lavkach3

1
import datetime
2
import uuid
3

4
from sqlalchemy import Column, DateTime, func, Uuid, BigInteger, ForeignKey, Sequence, text, types
5
from sqlalchemy.ext.declarative import declared_attr
6
from sqlalchemy.orm import mapped_column, Mapped
7
from sqlalchemy_utils.types import JSONType
8
from typing import Annotated, Optional
9

10
from sqlalchemy import DateTime
11
from sqlalchemy.ext.compiler import compiles
12
from sqlalchemy.sql import expression
13

14

15
# UTC now function on database server
16
class UTCNow(
17
    expression.FunctionElement  # type: ignore[name-defined]
18
):  # pylint: disable=too-many-ancestors
19
    type = DateTime()
20

21

22
@compiles(UTCNow, "postgresql")  # type: ignore[misc]
23
def pg_utcnow(  # type: ignore[no-untyped-def]
24
        element, compiler, **kw  # pylint: disable=unused-argument
25
) -> str:
26
    return "TIMEZONE('utc', CURRENT_TIMESTAMP)"
27

28
guid_primary_key = mapped_column(
29
        types.Uuid,
30
        primary_key=True,
31
        init=False,
32
        server_default=text("gen_random_uuid()")
33
    )
34
guid = mapped_column(
35
        types.Uuid,
36
        init=False,
37
        index=True,
38
        server_default=text("gen_random_uuid()")
39
    )
40
class VarsMixin:
41
    vars: Mapped[Optional[dict]] = mapped_column(JSONType)
42

43
class TimestampMixin:
44
    created_at: Mapped[datetime.datetime] = Column(
45
        DateTime(),
46
        nullable=False,
47
        server_default=UTCNow()
48
    )
49

50
    updated_at: Mapped[datetime.datetime] = Column(
51
        DateTime(),
52
        nullable=True,
53
        server_default=UTCNow(),
54
        server_onupdate=UTCNow(),
55
    )
56

57

58
class CompanyMixin:
59
    company_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("company.id"), index=True, nullable=False)
60

61
class CreatedEdited:
62
    created_by: Mapped[uuid.UUID] = mapped_column(Uuid, index=True, nullable=False)
63
    edited_by: Mapped[uuid.UUID] = mapped_column(Uuid, index=True, nullable=False)
64

65

66
class LsnMixin:
67
    @declared_attr
68
    def lsn_seq(cls):
69
        return Sequence(
70
            f'{cls.__tablename__}_lsn_seq'
71
        )
72

73
    @declared_attr
74
    def lsn(cls):
75
        return Column(
76
            BigInteger,
77
            getattr(cls, 'lsn_seq'),
78
            onupdate=getattr(cls, 'lsn_seq').next_value(), index=True
79
        )
80

81

82
class AllMixin(LsnMixin, CompanyMixin, TimestampMixin, VarsMixin):
83
    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, index=True, default=uuid.uuid4)