Добавление софт делитов в принтер

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/2b86076bf074_add_is_deleted_field_to_unionmember_.py

@@ -0,0 +1,32 @@
+"""Add is_deleted field to UnionMember table
+
+Revision ID: 2b86076bf074
+Revises: a68c6bb2972c
+Create Date: 2024-10-30 19:08:15.473750
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = '2b86076bf074'
+down_revision = 'a68c6bb2972c'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    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', 'is_deleted', nullable=False)
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('union_member', 'is_deleted')
+    op.alter_column('file', 'source', existing_type=sa.VARCHAR(), nullable=True)
+    # ### end Alembic commands ###

print_service/models/__init__.py

@@ -9,26 +9,24 @@
 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)
     union_number: Mapped[str] = mapped_column(String, nullable=True)
     student_number: Mapped[str] = mapped_column(String, nullable=True)
+    is_deleted: Mapped[bool] = mapped_column(Boolean, nullable=True, default=False)
 
     files: Mapped[list[File]] = relationship('File', back_populates='owner')
     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)
@@ -44,7 +42,11 @@ class File(Model):
     number_of_pages: Mapped[int] = Column(Integer)
     source: Mapped[str] = Column(String, default='unknown', nullable=False)
 
-    owner: Mapped[UnionMember] = relationship('UnionMember', back_populates='files')
+    owner: Mapped[UnionMember] = relationship(
+        'UnionMember',
+        primaryjoin="and_(File.owner_id==UnionMember.id, not_(UnionMember.is_deleted))",
+        back_populates='files',
+    )
     print_facts: Mapped[list[PrintFact]] = relationship('PrintFact', back_populates='file')
 
     @property
@@ -79,14 +81,18 @@ 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)
     owner_id: Mapped[int] = Column(Integer, ForeignKey('union_member.id'), nullable=False)
     created_at: Mapped[datetime] = Column(DateTime, nullable=False, default=datetime.utcnow)
 
-    owner: Mapped[UnionMember] = relationship('UnionMember', back_populates='print_facts')
+    owner: Mapped[UnionMember] = relationship(
+        'UnionMember',
+        primaryjoin="and_(PrintFact.owner_id == UnionMember.id, not_(UnionMember.is_deleted))",
+        back_populates='print_facts',
+    )
     file: Mapped[File] = relationship('File', back_populates='print_facts')
     sheets_used: Mapped[int] = Column(Integer)

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/file.py

@@ -113,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(
@@ -123,6 +123,7 @@ async def send(inp: SendInput, settings: Settings = Depends(get_settings)):
         ),
         func.upper(UnionMember.surname) == inp.surname.upper(),
     ).one_or_none()
+
     if not user:
         raise NotInUnion()
     try:
@@ -174,11 +175,12 @@ 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()
     )
+
     if not file_model:
         await file.close()
         raise PINNotFound(pin)
@@ -244,11 +246,12 @@ 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(options)
     if not file_model:
         raise PINNotFound(pin)

print_service/routes/user.py

@@ -40,9 +40,7 @@ class UpdateUserList(BaseModel):
 @router.get(
     '/is_union_member',
     status_code=202,
-    responses={
-        404: {'detail': 'User not found'},
-    },
+    responses={404: {'detail': 'User not found'}},
 )
 async def check_union_member(
     surname: constr(strip_whitespace=True, to_upper=True, min_length=1),
@@ -51,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(
@@ -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_(
@@ -111,15 +109,19 @@ def update_list(
         )
 
         if db_user:
-            db_user.surname = user.username
-            db_user.union_number = user.union_number
-            db_user.student_number = user.student_number
+            if db_user.is_deleted:
+                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(
                     surname=user.username,
                     union_number=user.union_number,
                     student_number=user.student_number,
+                    is_deleted=False,
                 )
             )
         db.session.flush()

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,11 +57,12 @@ 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()
     )
+
     if len(pin) != len(files):
         raise FileNotFound(len(pin) - len(files))
 

tests/test_routes/conftest.py

@@ -16,10 +16,23 @@ 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 = UnionMember.query(session=dbsession).filter(UnionMember.id == 42).one_or_none()
+    db_user.is_deleted = True
     dbsession.commit()
 
 
@@ -32,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()
 
 
@@ -60,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,14 +36,26 @@ 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)
     dbsession.delete(db_file2)
     dbsession.commit()
 
 
+def test_post_is_deleted(client, union_member_user, add_is_deleted_flag):
+    body = {
+        "surname": union_member_user['surname'],
+        "number": union_member_user['union_number'],
+        "filename": "filename.pdf",
+        "source": "webapp",
+        "options": {"pages": "", "copies": 1, "two_sided": False},
+    }
+    res = client.post(url, data=json.dumps(body))
+    assert res.status_code == status.HTTP_403_FORBIDDEN
+
+
 def test_post_unauthorized_user(client):
     body = {
         "surname": 'surname',