Fixed mypy complaints

Author: MrMatAP

.idea/runConfigurations/db_revision.xml

@@ -23,10 +23,11 @@
 
 from .config import Config
 app_config = Config.from_json_file()
+from sqlalchemy.orm import DeclarativeBase
 
-from sqlalchemy.orm import declarative_base
-Base = declarative_base()
+class ORMBase(DeclarativeBase):
+    pass
 
 
-class BaseSchema(BaseModel):
+class SchemaBase(BaseModel):
     model_config = {'from_attributes': True}

.idea/runConfigurations/db_upgrade__head_.xml

@@ -20,9 +20,8 @@
 #  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 #  SOFTWARE.
 
-from typing import Union
 from fastapi import APIRouter, Depends, status, HTTPException
-from fastapi.responses import JSONResponse, Response
+from fastapi.responses import Response
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 
@@ -49,7 +48,10 @@
 async def get_resources(session: Session = Depends(get_db)) -> ResourceListSchema:
     try:
         resources = session.query(Resource).all()
-        return ResourceListSchema(resources=resources)
+        return ResourceListSchema(resources=[ResourceSchema(uid=r.uid,
+                                                            name=r.name,
+                                                            owner_uid=r.owner_uid)
+                                             for r in resources])
     except SQLAlchemyError as e:
         # Handle the error appropriately, maybe raise an HTTPException
         raise HTTPException(status_code=500, detail="A database error occurred") from e
@@ -86,12 +88,7 @@ async def get_resource(uid: str,
              name='create_resource',
              summary='Create a resource',
              description='Create a resource owned by the authenticated user',
-             responses={
-                 status.HTTP_201_CREATED: {
-                     'description': 'The resource was created',
-                     'model': ResourceSchema
-                 },
-             })
+             response_model=ResourceSchema)
 async def create_resource(data: ResourceInputSchema,
                           response: Response,
                           session: Session = Depends(get_db)):
@@ -100,9 +97,7 @@ async def create_resource(data: ResourceInputSchema,
         session.add(resource)
         session.commit()
         response.status_code = 201
-        return ResourceSchema(uid=resource.uid,
-                              owner_uid=resource.owner_uid,
-                              name=resource.name)
+        return resource
     except SQLAlchemyError as e:
         raise HTTPException(status_code=500, detail="A database error occurred") from e
 
@@ -120,7 +115,8 @@ async def create_resource(data: ResourceInputSchema,
                     'description': 'The resource was not found',
                     'model': StatusSchema
                 }
-            })
+            },
+            response_model=ResourceSchema)
 async def modify_resource(uid: str,
                           data: ResourceInputSchema,
                           response: Response,
@@ -133,9 +129,7 @@ async def modify_resource(uid: str,
         resource.name = data.name
         session.add(resource)
         session.commit()
-        return ResourceSchema(uid=resource.uid,
-                              owner_uid=resource.owner_uid,
-                              name=resource.name)
+        return resource
     except SQLAlchemyError as e:
         raise HTTPException(status_code=500, detail="A database error occurred") from e
 
@@ -176,10 +170,10 @@ async def remove_resource(uid: str,
             summary='List all known owners',
             description='Returns all currently known owners and their metadata',
             response_model=OwnerListSchema)
-async def get_owners(db: Session = Depends(get_db)):
+async def get_owners(session: Session = Depends(get_db)):
     try:
-        owners = db.query(Owner).all()
-        return OwnerListSchema(owners=owners)
+        owners = session.query(Owner).all()
+        return OwnerListSchema(owners=[OwnerSchema(uid=o.uid, name=o.name) for o in owners])
     except SQLAlchemyError as e:
         # Handle the error appropriately, maybe raise an HTTPException
         raise HTTPException(status_code=500, detail="A database error occurred") from e
@@ -196,9 +190,10 @@ async def get_owners(db: Session = Depends(get_db)):
                 },
                 status.HTTP_200_OK: {
                     'description': 'The requested owner',
-                    'model': ResourceSchema
+                    'model': OwnerSchema
                 }
-            })
+            },
+            response_model=OwnerSchema)
 async def get_owner(uid: str,
                     response: Response,
                     session: Session = Depends(get_db)):
@@ -216,9 +211,7 @@ async def get_owner(uid: str,
              name='create_owner',
              summary='Create a owner',
              description='Create a owner',
-             responses={
-                 status.HTTP_201_CREATED: {'description': 'The owner was created', 'model': OwnerSchema}
-             })
+             response_model=OwnerSchema)
 async def create_owner(data: OwnerInputSchema,
                        response: Response,
                        session: Session = Depends(get_db)):
@@ -227,7 +220,7 @@ async def create_owner(data: OwnerInputSchema,
         session.add(owner)
         session.commit()
         response.status_code = 201
-        return OwnerSchema(uid=owner.uid, name=owner.name)
+        return owner
     except SQLAlchemyError as e:
         # Handle the error appropriately, maybe raise an HTTPException
         raise HTTPException(status_code=500, detail="A database error occurred") from e
@@ -246,7 +239,8 @@ async def create_owner(data: OwnerInputSchema,
                     'description': 'The owner was not found',
                     'model': StatusSchema
                 }
-            })
+            },
+            response_model=OwnerSchema)
 async def modify_owner(uid: str,
                        data: OwnerInputSchema,
                        response: Response,
@@ -259,7 +253,7 @@ async def modify_owner(uid: str,
         owner.name = data.name
         session.add(owner)
         session.commit()
-        return OwnerSchema(uid=owner.uid, name=owner.name)
+        return owner
     except SQLAlchemyError as e:
         raise HTTPException(status_code=500, detail="A database error occurred") from e
 

src/mrmat_python_api_fastapi/__init__.py

@@ -21,28 +21,28 @@
 #  SOFTWARE.
 
 import uuid
-from sqlalchemy import ForeignKey, Column, String, UniqueConstraint, UUID
-from sqlalchemy.orm import relationship
+from sqlalchemy import ForeignKey, String, UniqueConstraint, UUID
+from sqlalchemy.orm import relationship, Mapped, mapped_column
 
-from mrmat_python_api_fastapi import Base
+from mrmat_python_api_fastapi import ORMBase
 
 
-class Owner(Base):
+class Owner(ORMBase):
     __tablename__ = 'owners'
     __schema__ = 'mrmat-python-api-fastapi'
-    #uid = Column(BigInteger().with_variant(Integer, 'sqlite'), primary_key=True)
-    uid = Column(String, primary_key=True, default=str(uuid.uuid4()))
-    client_id = Column(String(255), nullable=False, unique=True)
-    name = Column(String(255), nullable=False)
-    resources = relationship('Resource', back_populates='owner')
+    uid: Mapped[str] = mapped_column(String, primary_key=True, default=str(uuid.uuid4()))
 
+    client_id: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    resources: Mapped[list["Resource"]] = relationship('Resource', back_populates='owner')
 
-class Resource(Base):
+
+class Resource(ORMBase):
     __tablename__ = 'resources'
     __schema__ = 'mrmat-python-api-fastapi'
-    uid = Column(String, primary_key=True, default=str(uuid.uuid4()))
-    owner_uid = Column(String, ForeignKey('owners.uid'), nullable=False)
-    name = Column(String(255), nullable=False)
+    uid: Mapped[str] = mapped_column(String, primary_key=True, default=str(uuid.uuid4()))
+    owner_uid: Mapped[str] = mapped_column(String, ForeignKey('owners.uid'), nullable=False)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
 
-    owner = relationship('Owner', back_populates='resources')
-    UniqueConstraint('owner_uid', 'name', name='no_duplicate_names_per_owner')
+    owner: Mapped["Owner"] = relationship('Owner', back_populates='resources')
+    __table_args__ = (UniqueConstraint('owner_uid', 'name', name='no_duplicate_names_per_owner'),)

src/mrmat_python_api_fastapi/apis/platform/v1/api.py

@@ -22,25 +22,24 @@
 
 import typing
 
-from mrmat_python_api_fastapi import BaseSchema
+from mrmat_python_api_fastapi import SchemaBase
 
 
-class OwnerInputSchema(BaseSchema):
+class OwnerInputSchema(SchemaBase):
     name: str
 
 class OwnerSchema(OwnerInputSchema):
     uid: str
 
-class OwnerListSchema(BaseSchema):
-    owners: typing.List[OwnerSchema]
+class OwnerListSchema(SchemaBase):
+    owners: typing.Sequence[OwnerSchema]
 
-class ResourceInputSchema(BaseSchema):
+class ResourceInputSchema(SchemaBase):
     name: str
     owner_uid: str
 
 class ResourceSchema(ResourceInputSchema):
     uid: str
-    name: str
 
-class ResourceListSchema(BaseSchema):
-    resources: typing.List[ResourceSchema]
+class ResourceListSchema(SchemaBase):
+    resources: typing.Sequence[ResourceSchema]

src/mrmat_python_api_fastapi/apis/platform/v1/db.py

@@ -25,7 +25,7 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
 
-from mrmat_python_api_fastapi import app_config, Base
+from mrmat_python_api_fastapi import app_config, ORMBase
 
 
 @lru_cache
@@ -34,6 +34,6 @@ def get_db() -> Session:
         engine = create_engine(url=app_config.db_url, connect_args={'check_same_thread': False})
     else:
         engine = create_engine(url=app_config.db_url)
-    session_local = sessionmaker(bind=engine)
-    Base.metadata.create_all(bind=engine)
+    session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    ORMBase.metadata.create_all(bind=engine)
     return session_local()

src/mrmat_python_api_fastapi/apis/platform/v1/schema.py

@@ -24,7 +24,7 @@
 import pytest
 import fastapi.testclient
 
-from mrmat_python_api_fastapi import app_config, Base
+from mrmat_python_api_fastapi import app_config, ORMBase
 from mrmat_python_api_fastapi.app import app
 from mrmat_python_api_fastapi.db import get_db
 
@@ -46,5 +46,5 @@ def client(test_db_path) -> fastapi.testclient.TestClient:
     app_config.db_url = f'sqlite:///{test_db_path}'
     session = get_db()
     with session.begin():
-        Base.metadata.create_all(session.bind)
+        ORMBase.metadata.create_all(session.bind)
     return fastapi.testclient.TestClient(app)