upgraded to fully typed sqlalchemy

compendium_v2/db/model.py

-import enum
 import logging
-from typing import Any
+from decimal import Decimal
+from enum import Enum
+from typing import Optional
+from typing_extensions import Annotated
 
-from sqlalchemy import Column, Enum, Integer, MetaData, Numeric, String
-from sqlalchemy.orm import relationship, declarative_base
+from sqlalchemy import MetaData, String
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 from sqlalchemy.schema import ForeignKey
 
+
 logger = logging.getLogger(__name__)
 
 convention = {
@@ -18,37 +21,45 @@ convention = {
 
 metadata_obj = MetaData(naming_convention=convention)
 
-# https://github.com/python/mypy/issues/2477
-base_schema: Any = declarative_base(metadata=metadata_obj)
+str128 = Annotated[str, 128]
+int_pk = Annotated[int, mapped_column(primary_key=True)]
+int_pk_fkNREN = Annotated[int, mapped_column(ForeignKey("nren.id"), primary_key=True)]
+
+
+class Base(DeclarativeBase):
+    metadata = metadata_obj
+    type_annotation_map = {
+        str128: String(128),
+    }
 
 
-class NREN(base_schema):
+class NREN(Base):
     __tablename__ = 'nren'
-    id = Column(Integer, primary_key=True)
-    name = Column(String(128), nullable=False)
+    id: Mapped[int_pk]
+    name: Mapped[str128]
 
 
-class BudgetEntry(base_schema):
+class BudgetEntry(Base):
     __tablename__ = 'budgets'
-    nren_id = Column(Integer, ForeignKey(NREN.id), primary_key=True)
-    nren = relationship(NREN, lazy='joined')
-    year = Column(Integer, primary_key=True)
-    budget = Column(Numeric(asdecimal=False), nullable=False)
+    nren_id: Mapped[int_pk_fkNREN]
+    nren: Mapped[NREN] = relationship(lazy='joined')
+    year: Mapped[int_pk]
+    budget: Mapped[Decimal]
 
 
-class FundingSource(base_schema):
+class FundingSource(Base):
     __tablename__ = 'funding_source'
-    nren_id = Column(Integer, ForeignKey(NREN.id), primary_key=True)
-    nren = relationship(NREN, lazy='joined')
-    year = Column(Integer, primary_key=True)
-    client_institutions = Column(Numeric(asdecimal=False), nullable=False)
-    european_funding = Column(Numeric(asdecimal=False), nullable=False)
-    gov_public_bodies = Column(Numeric(asdecimal=False), nullable=False)
-    commercial = Column(Numeric(asdecimal=False), nullable=False)
-    other = Column(Numeric(asdecimal=False), nullable=False)
+    nren_id: Mapped[int_pk_fkNREN]
+    nren: Mapped[NREN] = relationship(lazy='joined')
+    year: Mapped[int_pk]
+    client_institutions: Mapped[Decimal]
+    european_funding: Mapped[Decimal]
+    gov_public_bodies: Mapped[Decimal]
+    commercial: Mapped[Decimal]
+    other: Mapped[Decimal]
 
 
-class FeeType(enum.Enum):
+class FeeType(Enum):
     flat_fee = "flat_fee"
     usage_based_fee = "usage_based_fee"
     combination = "combination"
@@ -56,45 +67,45 @@ class FeeType(enum.Enum):
     other = "other"
 
 
-class ChargingStructure(base_schema):
+class ChargingStructure(Base):
     __tablename__ = 'charging_structure'
-    nren_id = Column(Integer, ForeignKey(NREN.id), primary_key=True)
-    nren = relationship(NREN, lazy='joined')
-    year = Column(Integer, primary_key=True)
-    fee_type = Column('fee_type', Enum(FeeType, name="fee_type"), nullable=True)
+    nren_id: Mapped[int_pk_fkNREN]
+    nren: Mapped[NREN] = relationship(lazy='joined')
+    year: Mapped[int_pk]
+    fee_type: Mapped[Optional[FeeType]]
 
 
-class NrenStaff(base_schema):
+class NrenStaff(Base):
     __tablename__ = 'nren_staff'
-    nren_id = Column(Integer, ForeignKey(NREN.id), primary_key=True)
-    nren = relationship(NREN, lazy='joined')
-    year = Column(Integer, primary_key=True)
-    permanent_fte = Column(Numeric(asdecimal=False), nullable=False)
-    subcontracted_fte = Column(Numeric(asdecimal=False), nullable=False)
-    technical_fte = Column(Numeric(asdecimal=False), nullable=False)
-    non_technical_fte = Column(Numeric(asdecimal=False), nullable=False)
+    nren_id: Mapped[int_pk_fkNREN]
+    nren: Mapped[NREN] = relationship(lazy='joined')
+    year: Mapped[int_pk]
+    permanent_fte: Mapped[Decimal]
+    subcontracted_fte: Mapped[Decimal]
+    technical_fte: Mapped[Decimal]
+    non_technical_fte: Mapped[Decimal]
 
 
-class ParentOrganization(base_schema):
+class ParentOrganization(Base):
     __tablename__ = 'parent_organization'
-    nren_id = Column(Integer, ForeignKey(NREN.id), primary_key=True)
-    nren = relationship(NREN, lazy='joined')
-    year = Column(Integer, primary_key=True)
-    organization = Column(String(128), nullable=False)
+    nren_id: Mapped[int_pk_fkNREN]
+    nren: Mapped[NREN] = relationship(lazy='joined')
+    year: Mapped[int_pk]
+    organization: Mapped[str128]
 
 
-class SubOrganization(base_schema):
+class SubOrganization(Base):
     __tablename__ = 'sub_organization'
-    nren_id = Column(Integer, ForeignKey(NREN.id), primary_key=True)
-    nren = relationship(NREN, lazy='joined')
-    year = Column(Integer, primary_key=True)
-    organization = Column(String(128), primary_key=True)
-    role = Column(String(128), nullable=False)
+    nren_id: Mapped[int_pk_fkNREN]
+    nren: Mapped[NREN] = relationship(lazy='joined')
+    year: Mapped[int_pk]
+    organization: Mapped[str128] = mapped_column(primary_key=True)
+    role: Mapped[str128]
 
 
-class ECProject(base_schema):
+class ECProject(Base):
     __tablename__ = 'ec_project'
-    nren_id = Column(Integer, ForeignKey(NREN.id), primary_key=True)
-    nren = relationship(NREN, lazy='joined')
-    year = Column(Integer, primary_key=True)
-    project = Column(String(256), primary_key=True)
+    nren_id: Mapped[int_pk_fkNREN]
+    nren: Mapped[NREN] = relationship(NREN, lazy='joined')
+    year: Mapped[int_pk]
+    project: Mapped[str] = mapped_column(String(256), primary_key=True)

compendium_v2/migrations/README

-Single-database configuration for Flask.

compendium_v2/routes/funding.py

@@ -62,7 +62,7 @@ def funding_source_view() -> Any:
     def _extract_data(entry: model.FundingSource):
         return {
             'NREN': entry.nren.name,
-            'YEAR': int(entry.year),
+            'YEAR': entry.year,
             'CLIENT_INSTITUTIONS': float(entry.client_institutions),
             'EUROPEAN_FUNDING': float(entry.european_funding),
             'GOV_PUBLIC_BODIES': float(entry.gov_public_bodies),

compendium_v2/routes/staff.py

@@ -61,10 +61,10 @@ def staff_view() -> Any:
         return {
             'nren': entry.nren.name,
             'year': entry.year,
-            'permanent_fte': entry.permanent_fte,
-            'subcontracted_fte': entry.subcontracted_fte,
-            'technical_fte': entry.technical_fte,
-            'non_technical_fte': entry.non_technical_fte
+            'permanent_fte': float(entry.permanent_fte),
+            'subcontracted_fte': float(entry.subcontracted_fte),
+            'technical_fte': float(entry.technical_fte),
+            'non_technical_fte': float(entry.non_technical_fte)
         }
 
     with db.session_scope() as session:

compendium_v2/survey_db/model.py

 import logging
+from typing import List, Optional
 
-from typing import Any
-
-from sqlalchemy import Column, Integer, String
-from sqlalchemy.orm import declarative_base, relationship
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 from sqlalchemy.schema import ForeignKey
 
 logger = logging.getLogger(__name__)
 
-# https://github.com/python/mypy/issues/2477
-base_schema: Any = declarative_base()
+
+class Base(DeclarativeBase):
+    pass
 
 
-class Budgets(base_schema):
+class Budgets(Base):
     __tablename__ = 'budgets'
-    id = Column(Integer, primary_key=True)
-    budget = Column(String)
-    year = Column(Integer)
-    country_code = Column('country_code', String, ForeignKey('nrens.country_code'))
-    nren = relationship('Nrens', back_populates='budgets')
+    id: Mapped[int] = mapped_column(primary_key=True)
+    budget: Mapped[Optional[str]]
+    year: Mapped[Optional[int]]
+    country_code: Mapped[Optional[str]] = mapped_column(ForeignKey('nrens.country_code'))
+    nren: Mapped[Optional['Nrens']] = relationship(back_populates='budgets')
 
 
-class Nrens(base_schema):
+class Nrens(Base):
     __tablename__ = 'nrens'
-    id = Column(Integer, primary_key=True)
-    abbreviation = Column(String)
-    country_code = Column(String)
-    budgets = relationship('Budgets', back_populates='nren')
+    id: Mapped[int] = mapped_column(primary_key=True)
+    abbreviation: Mapped[Optional[str]]
+    country_code: Mapped[Optional[str]]
+    budgets: Mapped[List['Budgets']] = relationship(back_populates='nren')

test/conftest.py

@@ -40,7 +40,7 @@ def mocked_survey_db(mocker):
         connect_args={'check_same_thread': False},
         poolclass=StaticPool,
         echo=False)
-    survey_model.base_schema.metadata.create_all(engine)
+    survey_model.Base.metadata.create_all(engine)
     mocker.patch(
         'compendium_v2.survey_db._SESSION_MAKER',
         sessionmaker(bind=engine))
@@ -57,7 +57,7 @@ def mocked_db(mocker):
         connect_args={'check_same_thread': False},
         poolclass=StaticPool,
         echo=False)
-    model.base_schema.metadata.create_all(engine)
+    model.Base.metadata.create_all(engine)
     mocker.patch(
         'compendium_v2.db._SESSION_MAKER',
         sessionmaker(bind=engine))