Pydantic v2

migrations/env.py

@@ -40,7 +40,7 @@ def run_migrations_offline():
     script output.
 
     """
-    url = settings.DB_DSN
+    url = str(settings.DB_DSN)
     context.configure(
         url=url,
         target_metadata=target_metadata,
@@ -60,7 +60,7 @@ def run_migrations_online():
 
     """
     configuration = config.get_section(config.config_ini_section)
-    configuration['sqlalchemy.url'] = settings.DB_DSN
+    configuration['sqlalchemy.url'] = str(settings.DB_DSN)
     connectable = engine_from_config(
         configuration,
         prefix='sqlalchemy.',

print_service/base.py

@@ -1,15 +1,14 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 
 class Base(BaseModel):
     def __repr__(self) -> str:
         attrs = []
-        for k, v in self.__class__.schema().items():
+        for k, v in self.__class__.model_json_schema().items():
             attrs.append(f"{k}={v}")
         return "{}({})".format(self.__class__.__name__, ', '.join(attrs))
 
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True, extra="ignore")
 
 
 class StatusResponseModel(Base):

print_service/routes/admin.py

@@ -26,7 +26,7 @@ class RebootInput(BaseModel):
 class InstantCommandSender:
     def __init__(self, settings: Settings = None) -> None:
         settings = settings or get_settings()
-        self.redis: Redis = Redis.from_url(settings.REDIS_DSN)
+        self.redis: Redis = Redis.from_url(str(settings.REDIS_DSN))
 
     def update(self, terminal_token: str):
         terminal = self.redis.get(terminal_token)

print_service/routes/base.py

@@ -27,7 +27,9 @@
     docs_url=None if __version__ != 'dev' else '/docs',
     redoc_url=None,
 )
-app.add_middleware(DBSessionMiddleware, db_url=settings.DB_DSN, engine_args=dict(pool_pre_ping=True))
+app.add_middleware(
+    DBSessionMiddleware, db_url=str(settings.DB_DSN), engine_args=dict(pool_pre_ping=True)
+)
 
 app.add_middleware(
     CORSMiddleware,

print_service/routes/exc_handlers.py

@@ -36,7 +36,7 @@ async def too_large_size(req: starlette.requests.Request, exc: TooLargeSize):
             status="Error",
             message=f"{exc}",
             ru=f"Размер файла превышает максимально допустимый: {settings.MAX_SIZE}",
-        ).dict(),
+        ).model_dump(),
         status_code=413,
     )
 
@@ -48,7 +48,7 @@ async def too_many_pages(req: starlette.requests.Request, exc: TooManyPages):
             status="Error",
             message=f"{exc}",
             ru=f"Количество запрошенных страниц превышает допустимое число: {settings.MAX_PAGE_COUNT}",
-        ).dict(),
+        ).model_dump(),
         status_code=413,
     )
 
@@ -60,7 +60,7 @@ async def invalid_format(req: starlette.requests.Request, exc: TooManyPages):
             status="Error",
             message=f"{exc}",
             ru="Количество запрошенных страниц превышает их количество в файле",
-        ).dict(),
+        ).model_dump(),
         status_code=416,
     )
 
@@ -70,7 +70,7 @@ async def terminal_not_found_by_qr(req: starlette.requests.Request, exc: Termina
     return JSONResponse(
         content=StatusResponseModel(
             status="Error", message="Terminal not found by QR", ru="QR-код не найден"
-        ).dict(),
+        ).model_dump(),
         status_code=400,
     )
 
@@ -80,7 +80,7 @@ async def terminal_not_found_by_token(req: starlette.requests.Request, exc: Term
     return JSONResponse(
         content=StatusResponseModel(
             status="Error", message="Terminal not found by token", ru="Токен не найден"
-        ).dict(),
+        ).model_dump(),
         status_code=400,
     )
 
@@ -90,7 +90,7 @@ async def user_not_found(req: starlette.requests.Request, exc: UserNotFound):
     return JSONResponse(
         content=StatusResponseModel(
             status="Error", message="User not found", ru="Пользователь не найден"
-        ).dict(),
+        ).model_dump(),
         status_code=404,
     )
 
@@ -102,7 +102,7 @@ async def student_duplicate(req: starlette.requests.Request, exc: UnionStudentDu
             status="Error",
             message=f"{exc}",
             ru="Один или более пользователей в списке не являются уникальными",
-        ).dict(),
+        ).model_dump(),
         status_code=400,
     )
 
@@ -114,7 +114,7 @@ async def not_in_union(req: starlette.requests.Request, exc: NotInUnion):
             status="Error",
             message=f"{exc}",
             ru="Отсутствует членство в профсоюзе",
-        ).dict(),
+        ).model_dump(),
         status_code=403,
     )
 
@@ -126,15 +126,15 @@ async def generate_error(req: starlette.requests.Request, exc: PINGenerateError)
             status="Error",
             message=f"{exc}",
             ru="Ошибка генерации ПИН-кода",
-        ).dict(),
+        ).model_dump(),
         status_code=500,
     )
 
 
 @app.exception_handler(FileIsNotReceived)
 async def file_not_received(req: starlette.requests.Request, exc: FileIsNotReceived):
     return JSONResponse(
-        content=StatusResponseModel(status="Error", message=f"{exc}", ru="Файл не получен").dict(),
+        content=StatusResponseModel(status="Error", message=f"{exc}", ru="Файл не получен").model_dump(),
         status_code=400,
     )
 
@@ -144,7 +144,7 @@ async def pin_not_found(req: starlette.requests.Request, exc: PINNotFound):
     return JSONResponse(
         content=StatusResponseModel(
             status="Error", message=f"Pin {exc.pin} not found", ru="ПИН не найден"
-        ).dict(),
+        ).model_dump(),
         status_code=404,
     )
 
@@ -156,23 +156,25 @@ async def invalid_type(req: starlette.requests.Request, exc: InvalidType):
             status="Error",
             message=f"{exc}",
             ru=f"Неподдерживаемый формат файла. Допустимые: {', '.join(settings.CONTENT_TYPES)}",
-        ).dict(),
+        ).model_dump(),
         status_code=415,
     )
 
 
 @app.exception_handler(AlreadyUploaded)
 async def already_upload(req: starlette.requests.Request, exc: AlreadyUploaded):
     return JSONResponse(
-        content=StatusResponseModel(status="Error", message=f"{exc}", ru="Файл уже загружен").dict(),
+        content=StatusResponseModel(
+            status="Error", message=f"{exc}", ru="Файл уже загружен"
+        ).model_dump(),
         status_code=415,
     )
 
 
 @app.exception_handler(IsCorrupted)
 async def is_corrupted(req: starlette.requests.Request, exc: IsCorrupted):
     return JSONResponse(
-        content=StatusResponseModel(status="Error", message=f"{exc}", ru="Файл повреждён").dict(),
+        content=StatusResponseModel(status="Error", message=f"{exc}", ru="Файл повреждён").model_dump(),
         status_code=415,
     )
 
@@ -182,7 +184,7 @@ async def unprocessable_file_instance(req: starlette.requests.Request, exc: Unpr
     return JSONResponse(
         content=StatusResponseModel(
             status="Error", message=f"{exc}", ru="Необрабатываемый экземпляр файла"
-        ).dict(),
+        ).model_dump(),
         status_code=422,
     )
 
@@ -192,14 +194,16 @@ async def file_not_found(req: starlette.requests.Request, exc: FileNotFound):
     return JSONResponse(
         content=StatusResponseModel(
             status="Error", message=f"{exc.count} file(s) not found", ru="Файл не найден"
-        ).dict(),
+        ).model_dump(),
         status_code=404,
     )
 
 
 @app.exception_handler(IsNotUploaded)
 async def not_uploaded(req: starlette.requests.Request, exc: IsNotUploaded):
     return JSONResponse(
-        content=StatusResponseModel(status="Error", message=f"{exc}", ru="Файл не загружен").dict(),
+        content=StatusResponseModel(
+            status="Error", message=f"{exc}", ru="Файл не загружен"
+        ).model_dump(),
         status_code=415,
     )

print_service/routes/file.py

@@ -8,7 +8,7 @@
 from fastapi.exceptions import HTTPException
 from fastapi.params import Depends
 from fastapi_sqlalchemy import db
-from pydantic import Field, validator
+from pydantic import Field, field_validator
 from sqlalchemy import func, or_
 
 from print_service.base import StatusResponseModel
@@ -43,7 +43,7 @@ class PrintOptions(BaseModel):
     copies: int = Field(1, description='Количество копий для печати')
     two_sided: bool = Field(False, description='Включить печать с двух сторон листа')
 
-    @validator('pages', pre=True, always=True)
+    @field_validator('pages', mode='before')
     def validate_pages(cls, value: str):
         if not isinstance(value, str):
             raise ValueError('Value must be str')
@@ -75,7 +75,7 @@ class SendInput(BaseModel):
 
 
 class SendInputUpdate(BaseModel):
-    options: PrintOptions | None
+    options: PrintOptions | None = None
 
 
 class SendOutput(BaseModel):
@@ -234,7 +234,7 @@ async def update_file_options(
 
     Требует пин-код, полученный в методе POST `/file`. Обновлять настройки
     можно бесконечное количество раз. Можно изменять настройки по одной."""
-    options = inp.options.dict(exclude_unset=True)
+    options = inp.options.model_dump(exclude_unset=True)
     file_model = (
         db.session.query(FileModel)
         .filter(func.upper(FileModel.pin) == pin.upper())

print_service/routes/qrprint.py

@@ -3,11 +3,13 @@
 import random
 from asyncio import sleep
 from datetime import datetime, timedelta
+from typing import Set
 
 from fastapi import APIRouter, Header, HTTPException, WebSocket
 from fastapi_sqlalchemy import db
-from pydantic import conlist
+from pydantic import Field
 from redis import Redis
+from typing_extensions import Annotated
 
 from print_service.exceptions import FileNotFound, InvalidPageRequest, IsNotUploaded, TerminalQRNotFound
 from print_service.schema import BaseModel
@@ -22,13 +24,13 @@
 
 class InstantPrintCreate(BaseModel):
     qr_token: str
-    files: conlist(str, min_items=1, max_items=10, unique_items=True)
+    files: Annotated[Set[str], Field(min_length=1, max_length=10)]
 
 
 class InstantPrintSender:
     def __init__(self, settings: Settings = None) -> None:
         settings = settings or get_settings()
-        self.redis: Redis = Redis.from_url(settings.REDIS_DSN)
+        self.redis: Redis = Redis.from_url(str(settings.REDIS_DSN))
 
     def send(self, qr_token: str, files: list[str]):
         terminal = self.redis.get(qr_token)
@@ -47,7 +49,7 @@ class InstantPrintFetcher:
     def __init__(self, terminal_token: str, settings: Settings = None) -> None:
         self.terminal_token = terminal_token
         settings = settings or get_settings()
-        self.redis = Redis.from_url(settings.REDIS_DSN)
+        self.redis = Redis.from_url(str(settings.REDIS_DSN))
         self.ttl = settings.QR_TOKEN_TTL
         self.delay = settings.QR_TOKEN_DELAY
         self.symbols = settings.QR_TOKEN_SYMBOLS
@@ -90,7 +92,7 @@ async def __anext__(self):
 
 @router.post("")
 async def instant_print(options: InstantPrintCreate):
-    options.qr_token = options.qr_token.removeprefix(settings.QR_TOKEN_PREFIX)
+    options.qr_token = options.qr_token.removeprefix(str(settings.QR_TOKEN_PREFIX))
     if redis_conn.send(**options.dict()):
         return {'status': 'ok'}
     raise TerminalQRNotFound()
@@ -103,7 +105,7 @@ async def instant_print_terminal_connection(
 ):
     await websocket.accept()
     manager = InstantPrintFetcher(authorization.removeprefix("token "))
-    await websocket.send_text(json.dumps({"qr_token": settings.QR_TOKEN_PREFIX + manager.new_qr()}))
+    await websocket.send_text(json.dumps({"qr_token": str(settings.QR_TOKEN_PREFIX) + manager.new_qr()}))
     async for task in manager:
-        task['qr_token'] = settings.QR_TOKEN_PREFIX + task['qr_token']
+        task['qr_token'] = str(settings.QR_TOKEN_PREFIX) + task['qr_token']
         await websocket.send_text(json.dumps(task))

print_service/routes/user.py

@@ -5,7 +5,7 @@
 from fastapi import APIRouter, Depends
 from fastapi.exceptions import HTTPException
 from fastapi_sqlalchemy import db
-from pydantic import constr
+from pydantic import constr, validate_call
 from sqlalchemy import and_, func, or_
 
 from print_service import __version__
@@ -35,6 +35,8 @@ class UpdateUserList(BaseModel):
 
 
 # region handlers
+
+
 @router.get(
     '/is_union_member',
     status_code=202,
@@ -48,6 +50,7 @@ async def check_union_member(
     v: Optional[str] = __version__,
 ):
     """Проверяет наличие пользователя в списке."""
+    surname = surname.upper()
     user = db.session.query(UnionMember)
     if not settings.ALLOW_STUDENT_NUMBER:
         user = user.filter(UnionMember.union_number != None)

print_service/settings.py

@@ -4,7 +4,8 @@
 from typing import List
 
 from auth_lib.fastapi import UnionAuthSettings
-from pydantic import AnyUrl, BaseSettings, DirectoryPath, PostgresDsn, RedisDsn
+from pydantic import AnyUrl, ConfigDict, DirectoryPath, PostgresDsn, RedisDsn
+from pydantic_settings import BaseSettings
 
 
 class Settings(UnionAuthSettings, BaseSettings):
@@ -18,7 +19,7 @@ class Settings(UnionAuthSettings, BaseSettings):
     MAX_SIZE: int = 26214400  # Максимальный размер файла в байтах
     MAX_PAGE_COUNT: int = 50
     STORAGE_TIME: int = 7 * 24  # Время хранения файла в часах
-    STATIC_FOLDER: DirectoryPath | None
+    STATIC_FOLDER: DirectoryPath | None = None
 
     ALLOW_STUDENT_NUMBER: bool = False
 
@@ -36,8 +37,7 @@ class Settings(UnionAuthSettings, BaseSettings):
     QR_TOKEN_TTL: int = 30  # Show time of QR code in seconds
     QR_TOKEN_DELAY: int = 5  # How long QR code valid after hide in seconds
 
-    class Config:
-        env_file = '.env'
+    model_config = ConfigDict(env_file=".env", extra="allow")
 
 
 @lru_cache()

requirements.txt

@@ -13,3 +13,4 @@ auth_lib_profcomff[fastapi]
 redis
 PyPDF4
 logging-profcomff
+pydantic-settings

tests/conftest.py

@@ -25,6 +25,6 @@ def client(mocker):
 @pytest.fixture
 def dbsession() -> Session:
     settings = Settings()
-    engine = create_engine(settings.DB_DSN, pool_pre_ping=True)
+    engine = create_engine(str(settings.DB_DSN), pool_pre_ping=True)
     TestingSessionLocal = sessionmaker(bind=engine)
     yield TestingSessionLocal()