f

migrations/env.py

@@ -3,7 +3,7 @@
 from alembic import context
 from sqlalchemy import engine_from_config, pool
 
-from print_service.models import Model
+from print_service.models.base import Base
 from print_service.settings import get_settings
 
 
@@ -20,7 +20,7 @@
 # for 'autogenerate' support
 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
-target_metadata = Model.metadata
+target_metadata = Base.metadata
 
 # other values from the config, defined by the needs of env.py,
 # can be acquired:

migrations/versions/120e06710249_add_is_deleted_field_to_unionmember_.py → migrations/versions/2b86076bf074_add_is_deleted_field_to_unionmember_.py

@@ -1,8 +1,8 @@
 """Add is_deleted field to UnionMember table
 
-Revision ID: 120e06710249
+Revision ID: 2b86076bf074
 Revises: a68c6bb2972c
-Create Date: 2024-10-25 16:40:35.109586
+Create Date: 2024-10-30 19:08:15.473750
 
 """
 
@@ -11,7 +11,7 @@
 
 
 # revision identifiers, used by Alembic.
-revision = '120e06710249'
+revision = '2b86076bf074'
 down_revision = 'a68c6bb2972c'
 branch_labels = None
 depends_on = None
@@ -21,8 +21,6 @@ def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
     op.alter_column('file', 'source', existing_type=sa.VARCHAR(), nullable=False)
     op.add_column('union_member', sa.Column('is_deleted', sa.Boolean(), nullable=True))
-    op.execute(f'UPDATE "union_member" SET  is_deleted = False')
-    op.alter_column('union_member', sa.Column('is_deleted', sa.Boolean(), nullable=False))
     # ### end Alembic commands ###
 
 

print_service/exceptions.py

@@ -12,10 +12,6 @@ class TerminalTokenNotFound(ObjectNotFound):
     pass
 
 
-class UserIsDeleted(Exception):
-    pass
-
-
 class TerminalQRNotFound(ObjectNotFound):
     pass
 

print_service/models/__init__.py

@@ -9,14 +9,11 @@
 from sqlalchemy.sql.schema import ForeignKey
 from sqlalchemy.sql.sqltypes import Boolean
 
+from print_service.models.base import BaseDbModel
 
-@as_declarative()
-class Model:
-    pass
 
-
-class UnionMember(Model):
-    __tablename__ = 'union_member'
+class UnionMember(BaseDbModel):
+    #    __tablename__ = 'union_member'
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     surname: Mapped[str] = mapped_column(String, nullable=False)
@@ -28,8 +25,8 @@ class UnionMember(Model):
     print_facts: Mapped[list[PrintFact]] = relationship('PrintFact', back_populates='owner')
 
 
-class File(Model):
-    __tablename__ = 'file'
+class File(BaseDbModel):
+    #    __tablename__ = 'file'
 
     id: Mapped[int] = Column(Integer, primary_key=True)
     pin: Mapped[str] = Column(String, nullable=False)
@@ -84,8 +81,8 @@ def sheets_count(self) -> int | None:
             return len(self.flatten_pages) * self.option_copies
 
 
-class PrintFact(Model):
-    __tablename__ = 'print_fact'
+class PrintFact(BaseDbModel):
+    #    __tablename__ = 'print_fact'
 
     id: Mapped[int] = Column(Integer, primary_key=True)
     file_id: Mapped[int] = Column(Integer, ForeignKey('file.id'), nullable=False)

print_service/models/base.py

@@ -0,0 +1,29 @@
+import re
+
+import sqlalchemy
+from sqlalchemy import Integer, not_
+from sqlalchemy.orm import Mapped, Query, Session, as_declarative, declared_attr, mapped_column
+
+
+@as_declarative()
+class Base:
+    """Base class for all database entities"""
+
+    @declared_attr
+    def __tablename__(cls) -> str:  # pylint: disable=no-self-argument
+        """Generate database table name automatically.
+        Convert CamelCase class name to snake_case db table name.
+        """
+        return re.sub(r"(?<!^)(?=[A-Z])", "_", cls.__name__).lower()
+
+
+class BaseDbModel(Base):
+    __abstract__ = True
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+
+    @classmethod
+    def query(cls, session: Session, with_deleted: bool = False) -> Query:
+        objs = session.query(cls)
+        if not with_deleted and hasattr(cls, "is_deleted"):
+            objs = objs.filter(not_(cls.is_deleted))
+        return objs

print_service/routes/exc_handlers.py

@@ -20,7 +20,6 @@
     TooManyPages,
     UnionStudentDuplicate,
     UnprocessableFileInstance,
-    UserIsDeleted,
     UserNotFound,
 )
 from print_service.routes.base import app
@@ -76,16 +75,6 @@ async def terminal_not_found_by_qr(req: starlette.requests.Request, exc: Termina
     )
 
 
-@app.exception_handler(UserIsDeleted)
-async def user_is_deleted(req: starlette.requests.Request, exc: TerminalTokenNotFound):
-    return JSONResponse(
-        content=StatusResponseModel(
-            status="Error", message="User is deleted", ru="Пользователь удалён из базы"
-        ).model_dump(),
-        status_code=410,
-    )
-
-
 @app.exception_handler(TerminalTokenNotFound)
 async def terminal_not_found_by_token(req: starlette.requests.Request, exc: TerminalTokenNotFound):
     return JSONResponse(

print_service/routes/file.py

@@ -25,7 +25,6 @@
     TooLargeSize,
     TooManyPages,
     UnprocessableFileInstance,
-    UserIsDeleted,
     UserNotFound,
 )
 from print_service.models import File as FileModel
@@ -106,7 +105,6 @@ class ReceiveOutput(BaseModel):
     responses={
         403: {'model': StatusResponseModel, 'detail': 'User error'},
         500: {'model': StatusResponseModel, 'detail': 'PIN generate error'},
-        410: {'model': StatusResponseModel, 'detail': 'User is deleted'},
     },
     response_model=SendOutput,
 )
@@ -115,7 +113,7 @@ async def send(inp: SendInput, settings: Settings = Depends(get_settings)):
 
     Полученный пин-код можно использовать в методах POST и GET `/file/{pin}`.
     """
-    user = db.session.query(UnionMember)
+    user = UnionMember.query(session=db.session)
     if not settings.ALLOW_STUDENT_NUMBER:
         user = user.filter(UnionMember.union_number != None)
     user = user.filter(
@@ -125,9 +123,7 @@ async def send(inp: SendInput, settings: Settings = Depends(get_settings)):
         ),
         func.upper(UnionMember.surname) == inp.surname.upper(),
     ).one_or_none()
-    if user:
-        if user.is_deleted:
-            raise UserIsDeleted()
+
     if not user:
         raise NotInUnion()
     try:
@@ -179,7 +175,7 @@ async def upload_file(
     if file == ...:
         raise FileIsNotReceived()
     file_model = (
-        db.session.query(FileModel)
+        FileModel.query(session=db.session)
         .filter(func.upper(FileModel.pin) == pin.upper())
         .order_by(FileModel.created_at.desc())
         .one_or_none()
@@ -250,7 +246,7 @@ async def update_file_options(
     можно бесконечное количество раз. Можно изменять настройки по одной."""
     options = inp.options.model_dump(exclude_unset=True)
     file_model = (
-        db.session.query(FileModel)
+        FileModel.query(session=db.session)
         .filter(func.upper(FileModel.pin) == pin.upper())
         .order_by(FileModel.created_at.desc())
         .one_or_none()

print_service/routes/user.py

@@ -9,7 +9,7 @@
 from sqlalchemy import and_, func, or_
 
 from print_service import __version__
-from print_service.exceptions import UnionStudentDuplicate, UserIsDeleted, UserNotFound
+from print_service.exceptions import UnionStudentDuplicate, UserNotFound
 from print_service.models import UnionMember
 from print_service.schema import BaseModel
 from print_service.settings import get_settings
@@ -40,7 +40,7 @@ class UpdateUserList(BaseModel):
 @router.get(
     '/is_union_member',
     status_code=202,
-    responses={404: {'detail': 'User not found'}, 410: {'detail': 'User is deleted'}},
+    responses={404: {'detail': 'User not found'}},
 )
 async def check_union_member(
     surname: constr(strip_whitespace=True, to_upper=True, min_length=1),
@@ -49,7 +49,7 @@ async def check_union_member(
 ):
     """Проверяет наличие пользователя в списке."""
     surname = surname.upper()
-    user = db.session.query(UnionMember)
+    user = UnionMember.query(session=db.session)
     if not settings.ALLOW_STUDENT_NUMBER:
         user = user.filter(UnionMember.union_number != None)
     user: UnionMember = user.filter(
@@ -59,9 +59,7 @@ async def check_union_member(
         ),
         func.upper(UnionMember.surname) == surname,
     ).one_or_none()
-    if user:
-        if user.is_deleted:
-            raise UserIsDeleted()
+
     if v == '1':
         return bool(user)
 
@@ -94,7 +92,7 @@ def update_list(
 
     for user in input.users:
         db_user: UnionMember = (
-            db.session.query(UnionMember)
+            UnionMember.query(session=db.session, with_deleted=True)
             .filter(
                 or_(
                     and_(
@@ -112,13 +110,11 @@ def update_list(
 
         if db_user:
             if db_user.is_deleted:
-                raise UserIsDeleted()
-
-        if db_user:
-            db_user.surname = user.username
-            db_user.union_number = user.union_number
-            db_user.student_number = user.student_number
-            db_user.is_deleted = False
+                raise UserNotFound
+            else:
+                db_user.surname = user.username
+                db_user.union_number = user.union_number
+                db_user.student_number = user.student_number
         else:
             db.session.add(
                 UnionMember(

print_service/utils/__init__.py

@@ -31,7 +31,7 @@ def generate_pin(session: Session):
     for i in range(15):
         pin = ''.join(random.choice(settings.PIN_SYMBOLS) for _ in range(settings.PIN_LENGTH))
         cnt = (
-            session.query(File)
+            File.query(session=session)
             .filter(
                 File.pin == pin,
                 File.created_at + timedelta(hours=settings.STORAGE_TIME) >= datetime.utcnow(),
@@ -57,7 +57,7 @@ def generate_filename(original_filename: str):
 def get_file(dbsession, pin: str or list[str]):
     pin = [pin.upper()] if isinstance(pin, str) else tuple(p.upper() for p in pin)
     files: list[FileModel] = (
-        dbsession.query(FileModel)
+        FileModel.query(session=dbsession)
         .filter(func.upper(FileModel.pin).in_(pin))
         .order_by(FileModel.created_at.desc())
         .all()

tests/test_routes/conftest.py

@@ -16,16 +16,22 @@ def union_member_user(dbsession):
     dbsession.add(UnionMember(**union_member))
     dbsession.commit()
     yield union_member
-    db_user = dbsession.query(UnionMember).filter(UnionMember.id == union_member['id']).one_or_none()
+    db_user = (
+        UnionMember.query(session=dbsession, with_deleted=True)
+        .filter(UnionMember.id == union_member['id'])
+        .one_or_none()
+    )
     assert db_user is not None
-    dbsession.query(PrintFact).filter(PrintFact.owner_id == union_member['id']).delete()
-    dbsession.query(UnionMember).filter(UnionMember.id == union_member['id']).delete()
+    PrintFact.query(session=dbsession).filter(PrintFact.owner_id == union_member['id']).delete()
+    UnionMember.query(session=dbsession, with_deleted=True).filter(
+        UnionMember.id == union_member['id']
+    ).delete()
     dbsession.commit()
 
 
 @pytest.fixture(scope='function')
 def add_is_deleted_flag(dbsession):
-    db_user = dbsession.query(UnionMember).filter(UnionMember.id == 42).one_or_none()
+    db_user = UnionMember.query(session=dbsession).filter(UnionMember.id == 42).one_or_none()
     db_user.is_deleted = True
     dbsession.commit()
 
@@ -39,12 +45,12 @@ def uploaded_file_db(dbsession, union_member_user, client):
         "options": {"pages": "", "copies": 1, "two_sided": False},
     }
     res = client.post('/file', json=body)
-    db_file = dbsession.query(File).filter(File.pin == res.json()['pin']).one_or_none()
+    db_file = File.query(session=dbsession).filter(File.pin == res.json()['pin']).one_or_none()
     yield db_file
-    file = dbsession.query(File).filter(File.pin == res.json()['pin']).one_or_none()
+    file = File.query(session=dbsession).filter(File.pin == res.json()['pin']).one_or_none()
     assert file is not None
-    dbsession.query(PrintFact).filter(PrintFact.file_id == file.id).delete()
-    dbsession.query(File).filter(File.pin == res.json()['pin']).delete()
+    PrintFact.query(session=dbsession).filter(PrintFact.file_id == file.id).delete()
+    File.query(session=dbsession).filter(File.pin == res.json()['pin']).delete()
     dbsession.commit()
 
 
@@ -67,8 +73,8 @@ def pin_pdf(dbsession, union_member_user, client):
     res = client.post('/file', json=body)
     pin = res.json()['pin']
     yield pin
-    file = dbsession.query(File).filter(File.pin == res.json()['pin']).one_or_none()
+    file = File.query(session=dbsession).filter(File.pin == res.json()['pin']).one_or_none()
     assert file is not None
-    dbsession.query(PrintFact).filter(PrintFact.file_id == file.id).delete()
-    dbsession.query(File).filter(File.pin == res.json()['pin']).delete()
+    PrintFact.query(session=dbsession).filter(PrintFact.file_id == file.id).delete()
+    File.query(session=dbsession).filter(File.pin == res.json()['pin']).delete()
     dbsession.commit()

tests/test_routes/test_file.py

@@ -25,7 +25,7 @@ def test_post_success(union_member_user, client, dbsession):
     }
     res = client.post(url, data=json.dumps(body))
     assert res.status_code == status.HTTP_200_OK
-    db_file = dbsession.query(File).filter(File.pin == res.json()['pin']).one_or_none()
+    db_file = File.query(session=dbsession).filter(File.pin == res.json()['pin']).one_or_none()
     assert db_file is not None
     assert db_file.source == 'webapp'
     body2 = {
@@ -36,7 +36,7 @@ def test_post_success(union_member_user, client, dbsession):
     }
     res2 = client.post(url, data=json.dumps(body2))
     assert res2.status_code == status.HTTP_200_OK
-    db_file2 = dbsession.query(File).filter(File.pin == res2.json()['pin']).one_or_none()
+    db_file2 = File.query(session=dbsession).filter(File.pin == res2.json()['pin']).one_or_none()
     assert db_file2 is not None
     assert db_file2.source == 'unknown'
     dbsession.delete(db_file)
@@ -53,7 +53,7 @@ def test_post_is_deleted(client, union_member_user, add_is_deleted_flag):
         "options": {"pages": "", "copies": 1, "two_sided": False},
     }
     res = client.post(url, data=json.dumps(body))
-    assert res.status_code == status.HTTP_410_GONE
+    assert res.status_code == status.HTTP_403_FORBIDDEN
 
 
 def test_post_unauthorized_user(client):