feat: use Mapped syntax for columns

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:

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.models import CustomBase
-
-from src.database import Base
 
 
-class Contact(Base):
+class Contact(CustomBase):
 	__tablename__ = "contact"
 
-	id = Column(Integer, primary_key=True)
+	id: Mapped[int] = mapped_column(primary_key=True)
-	email = Column(String)
+	email: Mapped[str]
-	first_name = Column(String)
+	first_name: Mapped[str]
-	last_name = Column(String)
+	last_name: Mapped[str]
-	phonenumber = Column(String)
+	phonenumber: Mapped[str]
-	vat_number = Column(String, default=None, nullable=True)
+	vat_number: Mapped[str | None] = mapped_column(default=None)
 
-	street_address = Column(String)
+	street_address : Mapped[str]
-	street_address_line_2 = Column(String)
+	street_address_line_2 : Mapped[str]
-	post_office_box_number = Column(String, default=None, nullable=True)
+	post_office_box_number: Mapped[str | None] = mapped_column(default=None)
-	locality = Column(String)  # Ie City
+	locality : Mapped[str]  # Ie City
-	country_code = Column(String)  # Eg GB
+	country_code : Mapped[str]  # Eg GB
-	address_region = Column(String, default=None, nullable=True)
+	address_region: Mapped[str | None] = mapped_column(default=None)
-	postal_code = Column(String)
+	postal_code : Mapped[str]
 
-	org_id = Column(
+	org_id: Mapped[int] = mapped_column(
-		Integer, ForeignKey("organisation.id", ondelete="CASCADE"), nullable=False
+		ForeignKey("organisation.id", ondelete="CASCADE"), nullable=False
 	)

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

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 import ForeignKey, UniqueConstraint
-from sqlalchemy.orm import relationship
+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)
+	id: Mapped[int] = mapped_column(primary_key=True)
-	resource = Column(String, nullable=False)
+	resource: Mapped[str]
-	action = Column(String, nullable=False)
+	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)
+	id: Mapped[int] = mapped_column(primary_key=True)
-	name = Column(String, nullable=False)
+	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(
+	group_id: Mapped[int] = mapped_column(
-		Integer, ForeignKey("group.id", ondelete="CASCADE"), primary_key=True
+		ForeignKey("group.id", ondelete="CASCADE"), primary_key=True
 	)
-	permission_id = Column(
+	permission_id: Mapped[int] = mapped_column(
-		Integer, ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True
+		ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True
 	)
 
 
-class UserGroups(Base):
+class UserGroups(CustomBase):
 	__tablename__ = "user_groups"
-	user_id = Column(
+	user_id: Mapped[int] = mapped_column(
-		Integer, ForeignKey("user.id", ondelete="CASCADE"), primary_key=True
+		ForeignKey("user.id", ondelete="CASCADE"), primary_key=True
 	)
-	group_id = Column(
+	group_id: Mapped[int] = mapped_column(
-		Integer, ForeignKey("group.id", ondelete="CASCADE"), primary_key=True
+		ForeignKey("group.id", ondelete="CASCADE"), primary_key=True
 	)
 
 
-class OrgPermissions(Base):
+class OrgPermissions(CustomBase):
 	__tablename__ = "org_permissions"
-	org_id = Column(
+	org_id: Mapped[int] = mapped_column(
-		Integer, ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True
+		ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True
 	)
-	permission_id = Column(
+	permission_id: Mapped[int] = mapped_column(
-		Integer, ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True
+		ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True
 	)

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,
+	}
+

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 import ForeignKey
-from sqlalchemy.orm import relationship
+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)
+	id: Mapped[int] = mapped_column(primary_key=True)
-	name = Column(String, unique=True)
+	name: Mapped[str]
-	status = Column(String, default="partial")
+	status: Mapped[str] = mapped_column(default="partial")
-	intake_questionnaire = Column(JSON)
+	intake_questionnaire: Mapped[dict[str, Any] | None]
 
-	root_user_id = Column(Integer, ForeignKey("user.id"))
+	root_user_id: Mapped[int] = mapped_column(ForeignKey("user.id"))
-
+	billing_contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id"))
-	billing_contact_id = Column(Integer, ForeignKey("contact.id"))
+	security_contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id"))
-	security_contact_id = Column(Integer, ForeignKey("contact.id"))
+	owner_contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id"))
-	owner_contact_id = Column(Integer, 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(
+	org_id: Mapped[int] = mapped_column(
-		Integer, ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True
+		ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True
-	)
-	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)

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, mapped_column, Mapped
-from sqlalchemy.orm import relationship
 
-from src.database import Base
+from src.models import CustomBase
 
 
-class Service(Base):
+class Service(CustomBase):
 	__tablename__ = "service"
 
-	id = Column(Integer, primary_key=True)
+	id: Mapped[int] = mapped_column(primary_key=True)
-	name = Column(String, unique=True)
+	name: Mapped[str] = mapped_column(unique=True)
-	api_key = Column(String, unique=True)
+	api_key: Mapped[str]
 
 	permission_rel = relationship("Permission", back_populates="service_rel")

src/user/models.py

@@ -12,33 +12,32 @@ Models:
 
 from collections import defaultdict
 
-from sqlalchemy import Column, Integer, String
+from sqlalchemy.orm import relationship, mapped_column, Mapped
-from sqlalchemy.orm import relationship
 
-from src.database import Base
+from src.models import CustomBase
 
 
-class User(Base):
+class User(CustomBase):
 	__tablename__ = "user"
 
-	id = Column(Integer, primary_key=True)
+	id: Mapped[int] = mapped_column(primary_key=True)
-	email = Column(String)
+	email: Mapped[str]
-	first_name = Column(String)
+	first_name: Mapped[str]
-	last_name = Column(String)
+	last_name: Mapped[str]
-	oidc_id = Column(String, index=True, unique=True)
+	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)

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)
+	CustomBase.metadata.drop_all(bind=engine)
-	Base.metadata.create_all(bind=engine)
+	CustomBase.metadata.create_all(bind=engine)
 	db = SessionLocal()
 	try:
 		_seed(db)  # extracted seeding logic into a plain function