SOFT-DELETES: all soft deletes

migrations/versions/a9c4bf6370de_soft_deletes.py

@@ -0,0 +1,45 @@
+"""soft-deletes
+
+Revision ID: a9c4bf6370de
+Revises: a68c6bb2972c
+Create Date: 2025-10-11 01:33:01.717991
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = 'a9c4bf6370de'
+down_revision = 'a68c6bb2972c'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column('file', sa.Column('is_deleted', sa.Boolean(), server_default='false', nullable=False))
+    op.alter_column('file', 'option_pages', existing_type=sa.VARCHAR(), nullable=False)
+    op.alter_column('file', 'option_copies', existing_type=sa.INTEGER(), nullable=False)
+    op.alter_column('file', 'option_two_sided', existing_type=sa.BOOLEAN(), nullable=False)
+    op.alter_column('file', 'number_of_pages', existing_type=sa.INTEGER(), nullable=False)
+    op.alter_column('file', 'source', existing_type=sa.VARCHAR(), nullable=False)
+    op.add_column(
+        'print_fact', sa.Column('is_deleted', sa.Boolean(), server_default='false', nullable=False)
+    )
+    op.alter_column('print_fact', 'sheets_used', existing_type=sa.INTEGER(), nullable=False)
+    op.add_column(
+        'union_member', sa.Column('is_deleted', sa.Boolean(), server_default='false', nullable=False)
+    )
+
+
+def downgrade():
+    op.drop_column('union_member', 'is_deleted')
+    op.alter_column('print_fact', 'sheets_used', existing_type=sa.INTEGER(), nullable=True)
+    op.drop_column('print_fact', 'is_deleted')
+    op.alter_column('file', 'source', existing_type=sa.VARCHAR(), nullable=True)
+    op.alter_column('file', 'number_of_pages', existing_type=sa.INTEGER(), nullable=True)
+    op.alter_column('file', 'option_two_sided', existing_type=sa.BOOLEAN(), nullable=True)
+    op.alter_column('file', 'option_copies', existing_type=sa.INTEGER(), nullable=True)
+    op.alter_column('file', 'option_pages', existing_type=sa.VARCHAR(), nullable=True)
+    op.drop_column('file', 'is_deleted')

print_service/base.py

@@ -1,17 +1,70 @@
-from pydantic import BaseModel, ConfigDict
+from __future__ import annotations
 
+import re
 
-class Base(BaseModel):
-    def __repr__(self) -> str:
+from sqlalchemy import not_
+from sqlalchemy.exc import NoResultFound
+from sqlalchemy.orm import Mapped, Query, Session, as_declarative, declared_attr
+
+from print_service.exceptions import ObjectNotFound
+
+
+@as_declarative()
+class Base:
+    @declared_attr
+    def __tablename__(cls) -> str:  # pylint: disable=no-self-argument
+        return re.sub(r"(?<!^)(?=[A-Z])", "_", cls.__name__).lower()
+
+    def __repr__(self):
         attrs = []
-        for k, v in self.__class__.model_json_schema().items():
-            attrs.append(f"{k}={v}")
-        return "{}({})".format(self.__class__.__name__, ', '.join(attrs))
+        for c in self.__table__.columns:
+            attrs.append(f"{c.name}={getattr(self, c.name)}")
+        return "{}({})".format(c.__class__.__name__, ', '.join(attrs))
+
+
+class BaseDbModel(Base):
+    __abstract__ = True
+
+    @classmethod
+    def create(cls, *, session: Session, **kwargs) -> BaseDbModel:
+        obj = cls(**kwargs)
+        session.add(obj)
+        session.flush()
+        return obj
+
+    @classmethod
+    def query(cls, *, with_deleted: bool = False, session: Session) -> Query:
+        objs = session.query(cls)
+        if not with_deleted and hasattr(cls, "is_deleted"):
+            objs = objs.filter(not_(cls.is_deleted))
+        return objs
+
+    @classmethod
+    def get(cls, id: int | str, *, with_deleted=False, session: Session) -> BaseDbModel:
+        objs = session.query(cls)
+        if not with_deleted and hasattr(cls, "is_deleted"):
+            objs = objs.filter(not_(cls.is_deleted))
+        try:
+            if hasattr(cls, "uuid"):
+                return objs.filter(cls.uuid == id).one()
+            return objs.filter(cls.id == id).one()
+        except NoResultFound:
+            raise ObjectNotFound(cls, id)
 
-    model_config = ConfigDict(from_attributes=True, extra="ignore")
+    @classmethod
+    def update(cls, id: int | str, *, session: Session, **kwargs) -> BaseDbModel:
+        obj = cls.get(id, session=session)
+        for k, v in kwargs.items():
+            setattr(obj, k, v)
 
+        session.flush()
+        return obj
 
-class StatusResponseModel(Base):
-    status: str
-    message: str
-    ru: str
+    @classmethod
+    def delete(cls, id: int | str, *, session: Session) -> None:
+        obj = cls.get(id, session=session)
+        if hasattr(obj, "is_deleted"):
+            obj.is_deleted = True
+        else:
+            session.delete(obj)
+        session.flush()

print_service/exceptions.py

@@ -1,33 +1,61 @@
+from typing import Type
+
 from print_service.settings import get_settings
 
 
 settings = get_settings()
 
 
-class ObjectNotFound(Exception):
-    pass
+class PrintAPIError(Exception):
+    eng: str
+    ru: str
+
+    def __init__(self, eng: str, ru: str) -> None:
+        self.eng = eng
+        self.ru = ru
+        super().__init__(eng)
+
+
+class ObjectNotFound(PrintAPIError):
+    def __init__(self, obj: type, obj_id_or_name: int | str):
+        super().__init__(
+            f"Object {obj.__name__} {obj_id_or_name=} not found",
+            f"Объект {obj.__name__} с идентификатором {obj_id_or_name} не найден",
+        )
+
+
+class AlreadyExists(PrintAPIError):
+    def __init__(self, obj: type, obj_id_or_name: int | str):
+        super().__init__(
+            f"Object {obj.__name__}, {obj_id_or_name=} already exists",
+            f"Объект {obj.__name__} с идентификатором {obj_id_or_name=} уже существует",
+        )
 
 
 class TerminalTokenNotFound(ObjectNotFound):
-    pass
+    def __init__(self, token_id: int | str):
+        super().__init__(type(self), token_id)
 
 
 class TerminalQRNotFound(ObjectNotFound):
-    pass
+    def __init__(self, qr_id: int | str):
+        super().__init__(type(self), qr_id)
 
 
 class PINNotFound(ObjectNotFound):
     def __init__(self, pin: str):
         self.pin = pin
+        super().__init__(type(self), pin)
 
 
 class UserNotFound(ObjectNotFound):
-    pass
+    def __init__(self, user_id: int | str):
+        super().__init__(type(self), user_id)
 
 
 class FileNotFound(ObjectNotFound):
-    def __init__(self, count: int):
-        self.count = count
+    def __init__(self, file_id: int | str):
+        super().__init__(type(self), file_id)
 
 
 class TooManyPages(Exception):

print_service/models/__init__.py

@@ -3,54 +3,55 @@
 import math
 from datetime import datetime
 
-from sqlalchemy import Column, DateTime, Integer, String
-from sqlalchemy.ext.declarative import as_declarative
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
-from sqlalchemy.sql.schema import ForeignKey
-from sqlalchemy.sql.sqltypes import Boolean
 
+from print_service.base import Base, BaseDbModel
 
-@as_declarative()
-class Model:
-    pass
 
+Model = Base
 
-class UnionMember(Model):
+
+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=False, default=False, server_default="false"
+    )
 
     files: Mapped[list[File]] = relationship('File', back_populates='owner')
     print_facts: Mapped[list[PrintFact]] = relationship('PrintFact', back_populates='owner')
 
 
-class File(Model):
+class File(BaseDbModel):
     __tablename__ = 'file'
 
-    id: Mapped[int] = Column(Integer, primary_key=True)
-    pin: Mapped[str] = Column(String, nullable=False)
-    file: Mapped[str] = Column(String, nullable=False)
-    owner_id: Mapped[int] = Column(Integer, ForeignKey('union_member.id'), nullable=False)
-    option_pages: Mapped[str] = Column(String)
-    option_copies: Mapped[int] = Column(Integer)
-    option_two_sided: Mapped[bool] = Column(Boolean)
-    created_at: Mapped[datetime] = Column(DateTime, nullable=False, default=datetime.utcnow)
-    updated_at: Mapped[datetime] = Column(
-        DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    pin: Mapped[str] = mapped_column(String, nullable=False)
+    file: Mapped[str] = mapped_column(String, nullable=False)
+    owner_id: Mapped[int] = mapped_column(Integer, ForeignKey('union_member.id'), nullable=False)
+    option_pages: Mapped[str] = mapped_column(String)
+    option_copies: Mapped[int] = mapped_column(Integer)
+    option_two_sided: Mapped[bool] = mapped_column(Boolean)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+    number_of_pages: Mapped[int] = mapped_column(Integer)
+    source: Mapped[str] = mapped_column(String, default='unknown', nullable=False)
+    is_deleted: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False, server_default="false"
     )
-    number_of_pages: Mapped[int] = Column(Integer)
-    source: Mapped[str] = Column(String, default='unknown', nullable=False)
 
     owner: Mapped[UnionMember] = relationship('UnionMember', back_populates='files')
     print_facts: Mapped[list[PrintFact]] = relationship('PrintFact', back_populates='file')
 
     @property
     def flatten_pages(self) -> list[int] | None:
-        '''Возвращает расширенный список из элементов списков внутренних целочисленных точек переданного множества отрезков
-        "1-5, 3, 2" --> [1, 2, 3, 4, 5, 3, 2]'''
         if self.number_of_pages is None:
             return None
         result = list()
@@ -63,9 +64,6 @@ def flatten_pages(self) -> list[int] | None:
 
     @property
     def sheets_count(self) -> int | None:
-        '''Возвращает количество элементов списков внутренних целочисленных точек переданного множества отрезков
-        "1-5, 3, 2" --> 7
-        P.S. 1, 2, 3, 4, 5, 3, 2 -- 7 чисел'''
         if self.number_of_pages is None:
             return None
         if not self.flatten_pages:
@@ -79,14 +77,17 @@ def sheets_count(self) -> int | None:
             return len(self.flatten_pages) * self.option_copies
 
 
-class PrintFact(Model):
+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)
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    file_id: Mapped[int] = mapped_column(Integer, ForeignKey('file.id'), nullable=False)
+    owner_id: Mapped[int] = mapped_column(Integer, ForeignKey('union_member.id'), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.now())
+    sheets_used: Mapped[int] = mapped_column(Integer)
+    is_deleted: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False, server_default="false"
+    )
 
     owner: Mapped[UnionMember] = relationship('UnionMember', back_populates='print_facts')
     file: Mapped[File] = relationship('File', back_populates='print_facts')
-    sheets_used: Mapped[int] = Column(Integer)