From 6397bd13160d1aafa3da1ccdb4fd9b60f7837481 Mon Sep 17 00:00:00 2001 From: irl Date: Sat, 20 Jun 2026 18:42:36 +0100 Subject: [PATCH] feat: use Mapped syntax for columns --- alembic/env.py | 4 +-- src/contact/models.py | 38 +++++++++++------------ src/database.py | 7 +---- src/iam/models.py | 62 +++++++++++++++++++------------------- src/models.py | 13 ++++++++ src/organisation/models.py | 43 ++++++++++++-------------- src/service/models.py | 13 ++++---- src/user/models.py | 25 ++++++++------- test/conftest.py | 8 ++--- 9 files changed, 108 insertions(+), 105 deletions(-) diff --git a/alembic/env.py b/alembic/env.py index 428e6f7..2e4486f 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -12,7 +12,7 @@ from src.organisation.models import Organisation, OrgUsers from src.user.models import User from src.service.models import Service from src.iam.models import Permission, Group, GroupPermissions, UserGroups -from src.database import Base +from src.models import CustomBase # this is the Alembic Config object, which provides # access to the values within the .ini file in use. @@ -27,7 +27,7 @@ if config.config_file_name is not None: # for 'autogenerate' support # from myapp import mymodel # target_metadata = mymodel.Base.metadata -target_metadata = Base.metadata +target_metadata = CustomBase.metadata # other values from the config, defined by the needs of env.py, # can be acquired: diff --git a/src/contact/models.py b/src/contact/models.py index d741bd2..9d0fdba 100644 --- a/src/contact/models.py +++ b/src/contact/models.py @@ -5,30 +5,30 @@ Models: - Contact: id[pk], email, first_name, last_name, phonenumber, vat_number street_address, street_address_line_2, post_office_box_number, address_locality, country_code, address_region, postal_code """ +from sqlalchemy import ForeignKey +from sqlalchemy.orm import mapped_column, Mapped -from sqlalchemy import Column, Integer, String, ForeignKey - -from src.database import Base +from src.models import CustomBase -class Contact(Base): +class Contact(CustomBase): __tablename__ = "contact" - id = Column(Integer, primary_key=True) - email = Column(String) - first_name = Column(String) - last_name = Column(String) - phonenumber = Column(String) - vat_number = Column(String, default=None, nullable=True) + id: Mapped[int] = mapped_column(primary_key=True) + email: Mapped[str] + first_name: Mapped[str] + last_name: Mapped[str] + phonenumber: Mapped[str] + vat_number: Mapped[str | None] = mapped_column(default=None) - street_address = Column(String) - street_address_line_2 = Column(String) - post_office_box_number = Column(String, default=None, nullable=True) - locality = Column(String) # Ie City - country_code = Column(String) # Eg GB - address_region = Column(String, default=None, nullable=True) - postal_code = Column(String) + street_address : Mapped[str] + street_address_line_2 : Mapped[str] + post_office_box_number: Mapped[str | None] = mapped_column(default=None) + locality : Mapped[str] # Ie City + country_code : Mapped[str] # Eg GB + address_region: Mapped[str | None] = mapped_column(default=None) + postal_code : Mapped[str] - org_id = Column( - Integer, ForeignKey("organisation.id", ondelete="CASCADE"), nullable=False + org_id: Mapped[int] = mapped_column( + ForeignKey("organisation.id", ondelete="CASCADE"), nullable=False ) diff --git a/src/database.py b/src/database.py index 3838098..8038905 100644 --- a/src/database.py +++ b/src/database.py @@ -5,10 +5,9 @@ Exports: - db_dependency - Base (sqlalchemy base model) """ - from typing import Annotated from sqlalchemy import create_engine, StaticPool -from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session +from sqlalchemy.orm import sessionmaker, Session from fastapi import Depends @@ -41,7 +40,3 @@ def get_db(): db_dependency = Annotated[Session, Depends(get_db)] - - -class Base(DeclarativeBase): - pass diff --git a/src/iam/models.py b/src/iam/models.py index c9ccc51..1860264 100644 --- a/src/iam/models.py +++ b/src/iam/models.py @@ -18,20 +18,20 @@ Models: - org_id[FK][PK], user_id[FK][PK], group_id[FK][PK] """ -from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint -from sqlalchemy.orm import relationship +from sqlalchemy import ForeignKey, UniqueConstraint +from sqlalchemy.orm import relationship, mapped_column, Mapped -from src.database import Base +from src.models import CustomBase -class Permission(Base): +class Permission(CustomBase): __tablename__ = "permission" - id = Column(Integer, primary_key=True) - resource = Column(String, nullable=False) - action = Column(String, nullable=False) + id: Mapped[int] = mapped_column(primary_key=True) + resource: Mapped[str] + action: Mapped[str] - service_id = Column(Integer, ForeignKey("service.id", ondelete="CASCADE")) + service_id: Mapped[int] = mapped_column(ForeignKey("service.id", ondelete="CASCADE")) __table_args__ = ( UniqueConstraint( @@ -46,10 +46,6 @@ class Permission(Base): "Service", back_populates="permission_rel", foreign_keys="Permission.service_id" ) - @property - def service_name(self): - return self.service_rel.name - group_rel = relationship( "Group", secondary="group_permissions", back_populates="permission_rel" ) @@ -58,13 +54,17 @@ class Permission(Base): "Organisation", secondary="org_permissions", back_populates="permission_rel" ) + @property + def service_name(self): + return self.service_rel.name -class Group(Base): + +class Group(CustomBase): __tablename__ = "group" - id = Column(Integer, primary_key=True) - name = Column(String, nullable=False) + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] - org_id = Column(Integer, ForeignKey("organisation.id", ondelete="CASCADE")) + org_id: Mapped[int] = mapped_column(ForeignKey("organisation.id", ondelete="CASCADE")) __table_args__ = ( UniqueConstraint( @@ -83,31 +83,31 @@ class Group(Base): ) -class GroupPermissions(Base): +class GroupPermissions(CustomBase): __tablename__ = "group_permissions" - group_id = Column( - Integer, ForeignKey("group.id", ondelete="CASCADE"), primary_key=True + group_id: Mapped[int] = mapped_column( + ForeignKey("group.id", ondelete="CASCADE"), primary_key=True ) - permission_id = Column( - Integer, ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True + permission_id: Mapped[int] = mapped_column( + ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True ) -class UserGroups(Base): +class UserGroups(CustomBase): __tablename__ = "user_groups" - user_id = Column( - Integer, ForeignKey("user.id", ondelete="CASCADE"), primary_key=True + user_id: Mapped[int] = mapped_column( + ForeignKey("user.id", ondelete="CASCADE"), primary_key=True ) - group_id = Column( - Integer, ForeignKey("group.id", ondelete="CASCADE"), primary_key=True + group_id: Mapped[int] = mapped_column( + ForeignKey("group.id", ondelete="CASCADE"), primary_key=True ) -class OrgPermissions(Base): +class OrgPermissions(CustomBase): __tablename__ = "org_permissions" - org_id = Column( - Integer, ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True + org_id: Mapped[int] = mapped_column( + ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True ) - permission_id = Column( - Integer, ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True + permission_id: Mapped[int] = mapped_column( + ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True ) diff --git a/src/models.py b/src/models.py index 87912aa..4f2e7b8 100644 --- a/src/models.py +++ b/src/models.py @@ -1,3 +1,16 @@ """ Global database models """ +from datetime import datetime +from typing import Any + +from sqlalchemy import DateTime, JSON +from sqlalchemy.orm import DeclarativeBase + + +class CustomBase(DeclarativeBase): + type_annotation_map = { + datetime: DateTime(timezone=True), + dict[str, Any]: JSON, + } + diff --git a/src/organisation/models.py b/src/organisation/models.py index c333c40..3813d69 100644 --- a/src/organisation/models.py +++ b/src/organisation/models.py @@ -13,26 +13,26 @@ Models: - owner_contact_rel: ORM relationship to Contact with owner_contact FK - OrgUsers: org_id[FK][PK], user_id[FK][PK] """ +from typing import Any -from sqlalchemy import Column, Integer, String, ForeignKey, JSON -from sqlalchemy.orm import relationship +from sqlalchemy import ForeignKey +from sqlalchemy.orm import relationship, Mapped, mapped_column -from src.database import Base +from src.models import CustomBase -class Organisation(Base): +class Organisation(CustomBase): __tablename__ = "organisation" - id = Column(Integer, primary_key=True) - name = Column(String, unique=True) - status = Column(String, default="partial") - intake_questionnaire = Column(JSON) + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] + status: Mapped[str] = mapped_column(default="partial") + intake_questionnaire: Mapped[dict[str, Any] | None] - root_user_id = Column(Integer, ForeignKey("user.id")) - - billing_contact_id = Column(Integer, ForeignKey("contact.id")) - security_contact_id = Column(Integer, ForeignKey("contact.id")) - owner_contact_id = Column(Integer, ForeignKey("contact.id")) + root_user_id: Mapped[int] = mapped_column(ForeignKey("user.id")) + billing_contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id")) + security_contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id")) + owner_contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id")) user_rel = relationship( "User", secondary="orgusers", back_populates="organisation_rel" @@ -41,10 +41,6 @@ class Organisation(Base): group_rel = relationship("Group", back_populates="org_rel") root_user_rel = relationship("User", foreign_keys="Organisation.root_user_id") - @property - def root_user_email(self): - return self.root_user_rel.email if self.root_user_rel else None - billing_contact_rel = relationship( "Contact", foreign_keys="Organisation.billing_contact_id" ) @@ -59,13 +55,14 @@ class Organisation(Base): "Permission", secondary="org_permissions", back_populates="org_rel" ) + @property + def root_user_email(self): + return self.root_user_rel.email if self.root_user_rel else None -class OrgUsers(Base): +class OrgUsers(CustomBase): __tablename__ = "orgusers" - org_id = Column( - Integer, ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True - ) - user_id = Column( - Integer, ForeignKey("user.id", ondelete="CASCADE"), primary_key=True + org_id: Mapped[int] = mapped_column( + ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True ) + user_id: Mapped[int] = mapped_column(ForeignKey("user.id", ondelete="CASCADE"), primary_key=True) diff --git a/src/service/models.py b/src/service/models.py index 414de5d..206b615 100644 --- a/src/service/models.py +++ b/src/service/models.py @@ -6,17 +6,16 @@ Models: - id[PK], name[U], api_key[U] """ -from sqlalchemy import Column, Integer, String -from sqlalchemy.orm import relationship +from sqlalchemy.orm import relationship, mapped_column, Mapped -from src.database import Base +from src.models import CustomBase -class Service(Base): +class Service(CustomBase): __tablename__ = "service" - id = Column(Integer, primary_key=True) - name = Column(String, unique=True) - api_key = Column(String, unique=True) + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] = mapped_column(unique=True) + api_key: Mapped[str] permission_rel = relationship("Permission", back_populates="service_rel") diff --git a/src/user/models.py b/src/user/models.py index 62c7ef6..4946d57 100644 --- a/src/user/models.py +++ b/src/user/models.py @@ -12,33 +12,32 @@ Models: from collections import defaultdict -from sqlalchemy import Column, Integer, String -from sqlalchemy.orm import relationship +from sqlalchemy.orm import relationship, mapped_column, Mapped -from src.database import Base +from src.models import CustomBase -class User(Base): +class User(CustomBase): __tablename__ = "user" - id = Column(Integer, primary_key=True) - email = Column(String) - first_name = Column(String) - last_name = Column(String) - oidc_id = Column(String, index=True, unique=True) + id: Mapped[int] = mapped_column(primary_key=True) + email: Mapped[str] + first_name: Mapped[str] + last_name: Mapped[str] + oidc_id: Mapped[str] = mapped_column(index=True, unique=True) organisation_rel = relationship( "Organisation", secondary="orgusers", back_populates="user_rel" ) - @property - def organisations(self): - return [{"name": org.name, "id": org.id} for org in self.organisation_rel] - group_rel = relationship( "Group", secondary="user_groups", back_populates="user_rel" ) + @property + def organisations(self): + return [{"name": org.name, "id": org.id} for org in self.organisation_rel] + @property def groups(self): result = defaultdict(list) diff --git a/test/conftest.py b/test/conftest.py index 417cd68..7ebf64c 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -14,16 +14,16 @@ from src.iam.models import Group, Permission, OrgPermissions from src.auth.service import get_current_user, get_dev_user from src.auth.dependencies import empty_su_list, get_super_admin_list, testing_su_list from src.main import app # inited FastAPI app -from src.database import engine, Base, get_db - +from src.database import engine, get_db +from models import CustomBase SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) @pytest.fixture() def db_session(): - Base.metadata.drop_all(bind=engine) - Base.metadata.create_all(bind=engine) + CustomBase.metadata.drop_all(bind=engine) + CustomBase.metadata.create_all(bind=engine) db = SessionLocal() try: _seed(db) # extracted seeding logic into a plain function