from __future__ import annotations
from sqlalchemy.dialects.postgresql.json import JSONB
import uuid
from sqlalchemy import Column, Text, ForeignKey
from sqlalchemy.orm import relationship
from onegov.core.orm import Base
from onegov.core.orm.mixins import TimestampMixin
from onegov.core.orm.types import UUID
from onegov.core.orm.types import JSON
from typing import Any
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from onegov.user import User
[docs]
class ImportLog(Base, TimestampMixin):
""" Logs the summary of a KUB data import attempt. """
[docs]
__tablename__ = 'pas_import_logs'
[docs]
id: Column[uuid.UUID] = Column(
UUID, # type: ignore[arg-type]
nullable=False,
primary_key=True,
default=uuid.uuid4
)
# user_id can be null if import is triggered by system/cron
[docs]
user_id: Column[uuid.UUID | None] = Column(
UUID, # type: ignore[arg-type]
ForeignKey('users.id'),
nullable=True
)
[docs]
user: relationship[User | None] = relationship('User')
# Store summary counts or potentially full details JSON
[docs]
details: Column[dict[str, Any]] = Column(JSON, nullable=False)
# 'completed' or 'failed'
[docs]
status: Column[str] = Column(Text, nullable=False)
# 'cli', 'upload', or 'automatic'
[docs]
import_type: Column[str] = Column(Text, nullable=False)
# This is the json response from the api, we store it to be able to
# reconstruct how certain imports were run. This will be useful to
# debug issues. But we don't want to store it every time, as these are
# several MBs of text and the sync cronjob runs regularly.
[docs]
people_source: Column[Any] = Column(JSONB, nullable=True)
[docs]
organizations_source: Column[Any] = Column(
JSONB, nullable=True
)
[docs]
memberships_source: Column[Any] = Column(JSONB, nullable=True)