users and status working

app/alchemy/__init__.py

@@ -0,0 +1,7 @@
+from .base import Base
+from .contact import Contact
+from .location import Country, LocationCode, Location
+from .printer import PrinterManufacturer, PrinterModel, Printer
+from .sap import SapNamePool
+from .status import Status
+from .user import User

app/alchemy/base.py

@@ -0,0 +1,44 @@
+from sqlalchemy.orm import DeclarativeBase, declared_attr, Mapped, mapped_column, relationship, add_mapped_attribute
+
+
+
+def to_snake_case(name: str) -> str:
+    return "".join(["_" + i.lower() if i.isupper() else i for i in name]).lstrip("_")
+
+
+
+def bidirectional_relationship(cls, foreign_table_cls):
+    """Create a bidirectional relationship between two table-classes."""
+
+    column_name = f"{to_snake_case(cls.__name__)}"
+    column_name = f"{column_name}es" if column_name.endswith("s") else f"{column_name}s"
+
+    add_mapped_attribute(
+        foreign_table_cls,
+        column_name,
+        relationship(cls,
+                     back_populates=foreign_table_cls.__tablename__,
+                     cascade="all, delete-orphan",
+                     collection_class=set,
+                     )
+    )
+
+    return relationship(foreign_table_cls.__name__, back_populates=column_name)
+
+
+class Base(DeclarativeBase):
+    __abstract__ = True
+
+    id: Mapped[int] = mapped_column(primary_key=True, sort_order=-1000000)
+
+    # noinspection PyMethodParameters
+    @declared_attr.directive
+    def __tablename__(cls) -> str:
+        """Default table name in Database is derived from Class Name"""
+        return to_snake_case(cls.__name__)
+
+
+# noinspection PyPep8Naming
+
+
+

app/alchemy/contact.py

@@ -0,0 +1,27 @@
+from sqlalchemy import String, Text, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, declared_attr
+
+from .base import Base, bidirectional_relationship
+# noinspection PyProtectedMember
+from .status import StatusForeignKey
+from .user import Versioned
+
+
+__all__ = ["Contact"]
+
+class Contact(StatusForeignKey, Versioned, Base):
+    """Contact"""
+
+    name: Mapped[str] = mapped_column(String(80), unique=True)
+    address: Mapped[str | None] = mapped_column(String(253))
+    notes: Mapped[str | None] = mapped_column(Text)
+
+
+class ContactForeignKey:
+    """Foreign Key Mixin for :py:class:`Contact`"""
+
+    contact_id: Mapped[int | None] = mapped_column(ForeignKey("contact.id"))
+
+    @declared_attr
+    def contact(cls) -> Mapped["Contact"]:
+        return bidirectional_relationship(cls, Contact)

app/alchemy/location.py

@@ -0,0 +1,88 @@
+from sqlalchemy import String, Text, ForeignKey, event, select, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, declared_attr, Session
+
+from .base import Base, bidirectional_relationship
+# noinspection PyProtectedMember
+from .contact import ContactForeignKey
+# noinspection PyProtectedMember
+from .status import StatusForeignKey
+from .user import Versioned, User
+
+__all__ = ["Country", "LocationCode", "Location"]
+
+
+class Country(StatusForeignKey, Versioned, Base):
+
+    iso: Mapped[str] = mapped_column(String(2), unique=True)
+    name: Mapped[str] = mapped_column(String(80))
+    notes: Mapped[str | None] = mapped_column(Text)
+
+
+class CountryForeignKey:
+
+    country_id: Mapped[int] = mapped_column(ForeignKey("country.id"))
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def country(cls) -> Mapped["Country"]:
+        return bidirectional_relationship(cls, Country)
+
+
+# noinspection PyUnusedLocal
+@event.listens_for(Country.__table__, "after_create")
+def initialize_country(target, connection, **kwargs):
+    with Session(connection) as session:
+        qsys = session.scalar(select(User).where(User.username == "QSYS"))
+        for kwargs in (
+                dict(iso="DE", name="Germany", status_id='A'),
+                dict(iso="IT", name="Italy", status_id='A'),
+                dict(iso="US", name="United States"),
+                dict(iso="CA", name="Canada"),
+                dict(iso="MX", name="Mexico"),
+                dict(iso="ES", name="Spain", status_id='A'),
+        ):
+            kwargs['_user__'] = qsys
+            session.add(Country(**kwargs))
+        session.commit()
+
+
+class LocationCode(CountryForeignKey, ContactForeignKey, StatusForeignKey, Versioned, Base):
+    """Location Code"""
+
+    code: Mapped[str] = mapped_column(String(8), unique=True)
+    description: Mapped[str] = mapped_column(String(256))
+    notes: Mapped[str | None] = mapped_column(Text)
+
+
+class LocationCodeForeignKey:
+    """Foreign Key Mixin for :py:class:`LocationCode`"""
+
+    location_code_id: Mapped[int] = mapped_column(ForeignKey("location_code.id"))
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def location_code(cls) -> Mapped["LocationCode"]:
+        return bidirectional_relationship(cls, LocationCode)
+
+
+class Location(LocationCodeForeignKey, ContactForeignKey, StatusForeignKey, Versioned, Base):
+    """Location"""
+
+    location: Mapped[str] = mapped_column(String(30))
+    description: Mapped[str] = mapped_column(String(256))
+    notes: Mapped[str | None] = mapped_column(Text)
+
+    __table_args__ = (
+        UniqueConstraint('location_code_id', location),
+    )
+
+
+class LocationForeignKey:
+    """Foreign Key Mixin for :py:class:`Location`"""
+
+    location_id: Mapped[int | None] = mapped_column(ForeignKey("location.id"))
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def location(cls) -> Mapped["Location"]:
+        return bidirectional_relationship(cls, Location)

app/alchemy/printer.py

@@ -0,0 +1,72 @@
+from sqlalchemy import String, Text, ForeignKey, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, declared_attr
+
+from .base import Base, bidirectional_relationship
+from .contact import ContactForeignKey
+from .location import LocationForeignKey
+from .status import StatusForeignKey
+from .user import Versioned
+
+
+class PrinterManufacturer(StatusForeignKey, Versioned, Base):
+    """Printer Manufacturer"""
+
+    code: Mapped[str] = mapped_column(String(10), unique=True)
+    description: Mapped[str] = mapped_column(String(256))
+    notes: Mapped[str | None] = mapped_column(Text)
+
+
+class PrinterManufacturerForeignKey:
+    """Foreign Key Mixin for :py:class:`PrinterManufacturer`"""
+
+    printer_manufacturer_id: Mapped[str] = mapped_column(ForeignKey("printer_manufacturer.id"))
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def printer_manufacturer(cls) -> Mapped["PrinterManufacturer"]:
+        return bidirectional_relationship(cls, PrinterManufacturer)
+
+
+class PrinterModel(PrinterManufacturerForeignKey, StatusForeignKey, Versioned, Base):
+    """Printer Model"""
+
+    code: Mapped[str] = mapped_column(String(20), unique=True)
+    description: Mapped[str] = mapped_column(String(256))
+    notes: Mapped[str | None] = mapped_column(Text)
+
+    __table_args__ = (
+        UniqueConstraint('printer_manufacturer_id', code),
+    )
+
+
+class PrinterModelForeignKey:
+    """Foreign Key Mixin for :py:class:`PrinterModel`"""
+
+    printer_model_id: Mapped[str] = mapped_column(ForeignKey("printer_model.id"))
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def printer_model(cls) -> Mapped["PrinterModel"]:
+        return bidirectional_relationship(cls, PrinterModel)
+
+
+class Printer(PrinterModelForeignKey, LocationForeignKey, ContactForeignKey, StatusForeignKey, Versioned, Base):
+    """Printer"""
+
+    name: Mapped[str] = mapped_column(String(63), unique=True)
+    description: Mapped[str] = mapped_column(String(256))
+    notes: Mapped[str | None] = mapped_column(Text)
+    dns_name: Mapped[str | None] = mapped_column(String(253))
+    port: Mapped[int | None]
+    location_detail: Mapped[str] = mapped_column(String(64), default='')
+
+
+class PrinterForeignKey:
+    """Foreign Key Mixin for :py:class:`Printer`"""
+
+    printer_id: Mapped[int] = mapped_column(ForeignKey("printer.id"))
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def printer(cls) -> Mapped["Printer"]:
+        return bidirectional_relationship(cls, Printer)

app/alchemy/sap.py

@@ -0,0 +1,24 @@
+from sqlalchemy import String, Text, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, declared_attr
+
+from .base import Base, bidirectional_relationship
+from .status import StatusForeignKey
+from .user import Versioned
+
+
+class SapNamePool(StatusForeignKey, Versioned, Base):
+    """SAP Printer Name Pool"""
+
+    code: Mapped[str] = mapped_column(String(12), primary_key=True, sort_order=-1000)
+    description: Mapped[str] = mapped_column(String(256))
+    notes: Mapped[str | None] = mapped_column(Text)
+
+
+class SapNamePoolForeignKey:
+    """Foreign Key Mixin for :py:class:`SapNamePool`"""
+
+    sap_name_pool_id: Mapped[str] = mapped_column(ForeignKey("sap_name_pool.id"))
+
+    @declared_attr
+    def sap_name_pool(cls) -> Mapped["SapNamePool"]:
+        return bidirectional_relationship(cls, SapNamePool)

app/alchemy/status.py

@@ -0,0 +1,42 @@
+from sqlalchemy import String, ForeignKey, event
+from sqlalchemy.orm import Mapped, mapped_column, declared_attr, relationship, Session
+
+from .base import Base
+
+__all__ = ["Status"]
+
+
+class Status(Base):
+    """Status of a record. Can be used in any table by using MixIn :class:`StatusForeignKey`."""
+
+    id: Mapped[str] = mapped_column(String(3), primary_key=True)
+    name: Mapped[str] = mapped_column(String(30), unique=True)
+
+
+class StatusForeignKey:
+    """Foreign Key Mixin for :py:class:`.Status`
+
+    By adding this mixin every record will get a status assigned.
+    """
+
+    status_id: Mapped[str] = mapped_column(ForeignKey("status.id"), default="N", sort_order=1000000)
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def status(cls) -> Mapped[Status]:
+        return relationship()
+
+
+# noinspection PyUnusedLocal
+@event.listens_for(Status.__table__, "after_create")
+def initialize_status(target, connection, **kwargs):
+    with Session(connection) as session:
+        for kwargs in (
+                dict(id="A", name="Active"),
+                dict(id="I", name="Inactive"),
+                dict(id="N", name="New"),
+                dict(id="PRE", name="Prepared"),
+                dict(id="X", name="eXcluded"),
+        ):
+            session.add(Status(**kwargs))
+        session.commit()

app/alchemy/user.py

@@ -0,0 +1,100 @@
+import sys
+from datetime import datetime
+
+from sqlalchemy import String, Text, ForeignKey, event, DateTime, select
+from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.orm import Mapped, mapped_column, declared_attr, relationship, Session
+from sqlalchemy.sql import expression
+
+from .base import Base
+from .status import StatusForeignKey
+
+__all__ = ['User', 'Versioned']
+
+
+# noinspection PyPep8Naming
+class utcnow(expression.FunctionElement):
+    type = DateTime()
+    inherit_cache = True
+
+
+# noinspection PyUnusedLocal
+@compiles(utcnow, "postgresql")
+def pg_utcnow(*args, **kwargs):
+    return "TIMEZONE('utc', CURRENT_TIMESTAMP)"
+
+
+# noinspection PyUnusedLocal
+@compiles(utcnow, "mssql")
+def ms_utcnow(*args, **kwargs):
+    return "GETUTCDATE()"
+
+
+# noinspection PyUnusedLocal
+@compiles(utcnow, "mysql")
+def my_utcnow(*args, **kwargs):
+    return "UTC_TIMESTAMP(6)"
+
+
+# noinspection PyUnusedLocal
+@compiles(utcnow, "mariadb")
+def maria_utcnow(*args, **kwargs):
+    return "UTC_TIMESTAMP(6)"
+
+
+# noinspection PyUnusedLocal
+@compiles(utcnow, "sqlite")
+def sqlite_utcnow(*args, **kwargs):
+    return "strftime('%Y-%m-%d %H:%M:%S')"
+
+
+class Versioned:
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def __versioned__(cls):
+        return {}
+
+    _created__: Mapped[datetime] = mapped_column(server_default=utcnow(), sort_order=sys.maxsize, default=None)
+    _updated__: Mapped[datetime | None] = mapped_column(onupdate=utcnow(), sort_order=sys.maxsize)
+    _user__id: Mapped[int | None] = mapped_column(ForeignKey("user.id"), sort_order=sys.maxsize)
+
+    # noinspection PyMethodParameters
+    @declared_attr
+    def _user__(cls) -> Mapped["User"]:
+        return relationship()
+
+
+class User(StatusForeignKey, Versioned, Base):
+    """User"""
+
+    username: Mapped[str] = mapped_column(String(253), unique=True)
+    name: Mapped[str] = mapped_column(String(253))
+    password: Mapped[str | None] = mapped_column(String(255))
+    ldap_name: Mapped[str | None] = mapped_column(String(255))
+    notes: Mapped[str | None] = mapped_column(Text)
+
+    def __repr__(self):
+        return f'User(id={self.id!r}, username={self.username!r} name={self.name!r}, notes={self.notes!r})'
+
+# noinspection PyUnusedLocal
+@event.listens_for(User.__table__, "after_create")
+def initialize_user(target, connection, **kwargs):
+    from routers.user import get_password_hash
+    with Session(connection) as session:
+        qsys = User(username="QSYS", name="System User", notes="internal processing", status_id='X')
+        session.add(qsys)
+        session.commit()
+        qsys = session.scalar(select(User).where(User.username == "QSYS"))
+        qsys._user__id=qsys.id
+        session.commit()
+        for kwargs in (
+                dict(username="CTM", name="Control-M", password=get_password_hash("secret"), notes="user for automation"),
+                dict(username="exde37c8", name="Bernhard Radermacher",
+                     password=get_password_hash("secret"),
+                     ldap_name="a0061806@kiongroup.com",
+                     ),
+        ):
+            kwargs.update(dict(status_id='A', _user__id=qsys.id))
+            session.add(User(**kwargs))
+        session.commit()