Add print fact and file number of pages (#45)

https://t.me/c/1758480664/28190/48403

migrations/versions/44494b133481_print_options.py

@@ -17,16 +17,12 @@
 
 
 def upgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
     op.add_column('file', sa.Column('option_pages', sa.String(), nullable=True))
     op.add_column('file', sa.Column('option_copies', sa.Integer(), nullable=True))
     op.add_column('file', sa.Column('option_two_sided', sa.Boolean(), nullable=True))
-    # ### end Alembic commands ###
 
 
 def downgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
     op.drop_column('file', 'option_two_sided')
     op.drop_column('file', 'option_copies')
     op.drop_column('file', 'option_pages')
-    # ### end Alembic commands ###

migrations/versions/686a37a323be_add_file_number_of_pages.py

@@ -0,0 +1,24 @@
+"""add_file_number_of_pages
+
+Revision ID: 686a37a323be
+Revises: 692fe4f50da7
+Create Date: 2023-04-29 19:38:02.676614
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = '686a37a323be'
+down_revision = '692fe4f50da7'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column('file', sa.Column('number_of_pages', sa.Integer(), nullable=True))
+
+
+def downgrade():
+    op.drop_column('file', 'number_of_pages')

migrations/versions/692fe4f50da7_upper_surnames.py

@@ -21,6 +21,4 @@ def upgrade():
 
 
 def downgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
     pass
-    # ### end Alembic commands ###

migrations/versions/f6fb6304fb74_add_print_fact.py

@@ -0,0 +1,39 @@
+"""add_print_fact
+
+Revision ID: f6fb6304fb74
+Revises: 686a37a323be
+Create Date: 2023-04-29 21:36:34.034457
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = 'f6fb6304fb74'
+down_revision = '686a37a323be'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'print_fact',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('file_id', sa.Integer(), nullable=False),
+        sa.Column('owner_id', sa.Integer(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ['file_id'],
+            ['file.id'],
+        ),
+        sa.ForeignKeyConstraint(
+            ['owner_id'],
+            ['union_member.id'],
+        ),
+        sa.PrimaryKeyConstraint('id'),
+    )
+
+
+def downgrade():
+    op.drop_table('print_fact')

print_service/models/__init__.py

@@ -23,6 +23,7 @@ class UnionMember(Model):
     student_number: Mapped[str] = mapped_column(String, nullable=True)
 
     files: Mapped[list[File]] = relationship('File', back_populates='owner')
+    print_facts: Mapped[list[PrintFact]] = relationship('PrintFact', back_populates='owner')
 
 
 class File(Model):
@@ -39,5 +40,19 @@ class File(Model):
     updated_at: Mapped[datetime] = Column(
         DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow
     )
+    number_of_pages: Mapped[int] = Column(Integer)
 
     owner: Mapped[UnionMember] = relationship('UnionMember', back_populates='files')
+    print_facts: Mapped[list[PrintFact]] = relationship('PrintFact', back_populates='file')
+
+
+class PrintFact(Model):
+    __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')
+    file: Mapped[File] = relationship('File', back_populates='print_facts')

print_service/routes/file.py

@@ -15,7 +15,7 @@
 from print_service.models import UnionMember
 from print_service.schema import BaseModel
 from print_service.settings import Settings, get_settings
-from print_service.utils import check_pdf_ok, generate_filename, generate_pin, get_file
+from print_service.utils import checking_for_pdf, generate_filename, generate_pin, get_file
 
 
 logger = logging.getLogger(__name__)
@@ -173,7 +173,10 @@ async def upload_file(
         if len(memory_file) > settings.MAX_SIZE:
             raise HTTPException(413, f'Content too large, {settings.MAX_SIZE} bytes allowed')
         await saved_file.write(memory_file)
-    if not check_pdf_ok(memory_file):
+    pdf_ok, number_of_pages = checking_for_pdf(memory_file)
+    file_model.number_of_pages = number_of_pages
+    db.session.commit()
+    if not pdf_ok:
         await aiofiles.os.remove(path)
         raise HTTPException(415, 'File corrupted')
     await file.close()

print_service/utils/__init__.py

@@ -12,6 +12,7 @@
 
 from print_service.models import File
 from print_service.models import File as FileModel
+from print_service.models import PrintFact
 from print_service.settings import Settings, get_settings
 
 
@@ -72,12 +73,24 @@ def get_file(dbsession, pin: str or list[str]):
                 },
             }
         )
+        file_model = PrintFact(file_id=f.id, owner_id=f.owner_id)
+        dbsession.add(file_model)
+        dbsession.commit()
     return result
 
 
-def check_pdf_ok(f: bytes):
+def checking_for_pdf(f: bytes) -> tuple[bool, int]:
+    """_summary_
+
+    Args:
+        f (bytes): file to check
+
+    Returns:
+        tuple[bool, int]: The first argument returns whether the file is a valid pdf.
+        The second argument returns the number of pages in the pdf document (0- if the check failed)
+    """
     try:
-        PdfFileReader(io.BytesIO(f))
-        return True
+        pdf_file = PdfFileReader(io.BytesIO(f))
+        return True, pdf_file.getNumPages()
     except Exception:
-        return False
+        return False, 0

tests/test_routes/conftest.py

@@ -1,9 +1,8 @@
 import os
-from unittest.mock import Mock
 
 import pytest
 
-from print_service.models import File, UnionMember
+from print_service.models import File, PrintFact, UnionMember
 
 
 @pytest.fixture(scope='function')
@@ -19,6 +18,7 @@ def union_member_user(dbsession):
     yield union_member
     db_user = dbsession.query(UnionMember).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()
     dbsession.commit()
 
@@ -34,6 +34,9 @@ def uploaded_file_db(dbsession, union_member_user, client):
     res = client.post('/file', json=body)
     db_file = dbsession.query(File).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()
+    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()
     dbsession.commit()
 
@@ -57,5 +60,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()
+    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()
     dbsession.commit()

tests/test_routes/test_file.py

@@ -1,16 +1,12 @@
-import asyncio
-import datetime
 import json
-import time
-from concurrent.futures import ThreadPoolExecutor
 
 import pytest
 from fastapi import HTTPException
 from starlette import status
 
 from print_service.models import File
 from print_service.settings import get_settings
-from print_service.utils import check_pdf_ok, generate_filename, get_file
+from print_service.utils import checking_for_pdf, get_file
 
 
 url = '/file'
@@ -91,8 +87,8 @@ def test_get_file_func_2_not_exists(dbsession, uploaded_file_os):
 
 
 def test_file_check():
-    assert check_pdf_ok(open("tests/test_routes/test_files/broken.pdf", "rb").read()) is False
-    assert check_pdf_ok(open("tests/test_routes/test_files/correct.pdf", "rb").read()) is True
+    assert checking_for_pdf(open("tests/test_routes/test_files/broken.pdf", "rb").read()) == (False, 0)
+    assert checking_for_pdf(open("tests/test_routes/test_files/correct.pdf", "rb").read()) == (True, 2)
 
 
 def test_upload_and_print_correct_pdf(pin_pdf, client):
@@ -130,9 +126,9 @@ def test_upload_and_print_encrypted_file(pin_pdf, client):
     fileName = 'tests/test_routes/test_files/encrypted.pdf'
     files = {'file': (f"{fileName}", open(f"{fileName}", 'rb'), "application/pdf")}
     res = client.post(f"{url}/{pin}", files=files)
-    assert res.status_code == status.HTTP_200_OK
+    assert res.status_code == status.HTTP_415_UNSUPPORTED_MEDIA_TYPE
     res2 = client.get(f"{url}/{pin}")
-    assert res2.status_code == status.HTTP_200_OK
+    assert res2.status_code == status.HTTP_415_UNSUPPORTED_MEDIA_TYPE
 
 
 def test_incorrect_filename(union_member_user, client, dbsession):