from datetime import datetime import enum from typing import Any, Dict, Generic, List, Literal, Optional, TypeVar from sqlalchemy.exc import MissingGreenlet FormPhaseRole = Literal["DFIP", "EXECUTOR_RF"] from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator T = TypeVar("T") ProjectTypeLiteral = Literal[ "Открытие ВСП", "Закрытие ВСП", "Переезд ВСП", "Реновация РФ", "Открытие УРМ", ] VspFormatLiteral = Literal[ "Флагманский", "Типовой", "Розничный", "МСБ", "Лёгкий", "Мини", "Розничный-киоск", "МБО", "Офис самообслуживания", "Другое", ] PlacementTypeLiteral = Literal[ "Собственность", "Аренда", "Субаренда", ] # def serialize_missing_greenlet(cls, obj): # result = {} # for attr in cls.model_fields: # try: # result[attr] = getattr(obj, attr) # except MissingGreenlet: # result[attr] = None # return result class ResponseBase(BaseModel): success: Optional[bool] = Field(True) message: Optional[str] = None class BaseSingleResponse(ResponseBase, Generic[T]): result: Optional[T] = None class BaseListResponse(ResponseBase, Generic[T]): result: Optional[list[T]] = None count: Optional[int] = None class UserBase(BaseModel): email: EmailStr = Field(..., examples=["user@example.com"]) username: str = Field( ..., min_length=3, max_length=64, pattern=r"^[A-Za-z0-9_.-]+$", examples=["user_1"], ) full_name: Optional[str] = Field( None, max_length=255, pattern=r"^[^<>]*$", examples=["Иван Иванов"], ) role_id: int = 3 model_config = ConfigDict(from_attributes=True) class UserCreate(UserBase): password: str | None = Field(default=None, examples=["pass123"]) load_orgs: bool = Field(default=False) model_config = ConfigDict( json_schema_extra={ "example": { "email": "user@example.com", "username": "user_1", "full_name": "Иван Иванов", "role_id": 3, "password": "pass123", "load_orgs": True, } } ) class UserUpdate(BaseModel): email: Optional[EmailStr] = Field(default=None, examples=["user@example.com"]) username: Optional[str] = Field( None, min_length=3, max_length=64, pattern=r"^[A-Za-z0-9_.-]+$", examples=["user_1"], ) full_name: Optional[str] = Field( None, max_length=255, pattern=r"^[^<>]*$", examples=["Иван Иванов"], ) role_id: Optional[int] = None is_active: Optional[bool] = True load_orgs: bool = Field(default=False) class UserInDB(UserBase): model_config = ConfigDict(from_attributes=True) id: int created_at: datetime updated_at: Optional[datetime] = None is_active: bool class User(UserInDB): pass class UserAdminListResponse(UserInDB): org_units: Optional[list["OrgUnitSchema"]] = None @model_validator(mode='before') @classmethod def safe_orm_handler(cls, data: Any) -> Any: if isinstance(data, dict): return data result = { "org_units": None, "id": data.id, "created_at": data.created_at, "updated_at": data.updated_at, "is_active": data.is_active, "email": data.email, "username": data.username, "full_name": data.full_name, "role_id": data.role_id, } try: result["org_units"] = [ ou for ou in data.org_units if ou.is_active ] except MissingGreenlet: pass return result class UserResponse(ResponseBase): result: Optional[UserInDB] = None class UserListResponse(ResponseBase): result: Optional[list[User]] = None class SSPBase(BaseModel): title: str is_ssp: bool = True is_active: bool = True class SSPInDB(SSPBase): model_config = ConfigDict(from_attributes=True) id: int class SSPIDList(BaseModel): ssp_ids: list[int] class UserSSPLinkBase(BaseModel): user_id: int ssp: list[SSPInDB] class UserSSPLinkResponse(ResponseBase): result: Optional[UserSSPLinkBase] = None class Token(BaseModel): access_token: str token_type: str = "bearer" expires_in: Optional[int] = None refresh_token: Optional[str] = None class LoginRequest(BaseModel): username: str password: str class RefreshRequest(BaseModel): refresh_token: str class FormTypeSchemaEnum(str, enum.Enum): FORM_1 = "FORM_1" FORM_2 = "FORM_2" FORM_3 = "FORM_3" FORM_4 = "FORM_4" class OrgUnitBaseSchema(BaseModel): title: str is_ssp: bool model_config = ConfigDict(from_attributes=True) class OrgUnitUpdateSchema(OrgUnitBaseSchema): title: Optional[str] = None is_ssp: Optional[bool] = None is_active: Optional[bool] = None model_config = ConfigDict(from_attributes=True) class OrgUnitResponseSchema(OrgUnitBaseSchema): id: int is_active: bool users_count: int | None = None model_config = ConfigDict(from_attributes=True) @model_validator(mode='before') @classmethod def safe_orm_handler(cls, data: Any) -> Any: if isinstance(data, dict): return data result = { "users_count": None, "id": data.id, "title": data.title, "is_ssp": data.is_ssp, "is_ssp": data.is_ssp, "is_active": data.is_active, } try: result["users_count"] = data.users_count except MissingGreenlet: pass return result class OrgUnitListSchema(OrgUnitResponseSchema): pass class OrgUnitSchema(OrgUnitResponseSchema): users: list[UserBase] | None = None @model_validator(mode='before') @classmethod def safe_orm_handler(cls, data: Any) -> Any: if isinstance(data, dict): return data result = super().safe_orm_handler(data) result["users"] = None try: result["users"] = data.users except MissingGreenlet: pass return result class BudgetFormResponse(BaseModel): id: int created_at: datetime created_by: int | None updated_at: datetime form_type_code: FormTypeSchemaEnum year: int org_unit_id: int org_unit: OrgUnitSchema | None = None model_config = ConfigDict(from_attributes=True) class FullSheetSchema(BaseModel): sheet_name: str direction: str | None = None class SheetFormTypeResponse(BaseModel): form_type: FormTypeSchemaEnum sheets: list[str] all_sheets: list[FullSheetSchema] | None = None class DirectionSchemaEnum(str, enum.Enum): SUPPORT = "Support" DEVELOPMENT = "Development" class SheetResponse(BaseModel): row_type: str depth: int data: dict sort_order: int | None = None class CellPatch(BaseModel): line_id: int column: str value: Any = None class CellsPatch(BaseModel): changes: list[CellPatch] class ExportBulkRequest(BaseModel): form_ids: list[int] = Field(min_length=1) skip_failed: bool = True class AddForm3LineBody(BaseModel): expense_item_id: int class UpdProjectBody(BaseModel): column: str value: Any = None class AddProjectBody(BaseModel): name: str = Field( ..., max_length=30, pattern=r"^[^+\-\/\\=&*\s]{1,30}$", ) year: int branch_id: int level: Literal["project", "program"] = "project" parent_id: Optional[int] = None ssp_id: Optional[int] = None project_type: Optional[ProjectTypeLiteral] = None vsp_format: Optional[VspFormatLiteral] = None placement_type: Optional[PlacementTypeLiteral] = None object_address: Optional[str] = None staff_count: Optional[int] = None total_area: Optional[float] = None class AddLineSchema(BaseModel): expense_item_id: Optional[int] = None item_id: Optional[str] = None section_code: Optional[str] = None direction: Optional[DirectionSchemaEnum] = None name: Optional[str] = None internal_order: Optional[str] = None vsp_id: Optional[int] = None project_id: Optional[int] = None justification: Optional[str] = None contract_number: Optional[str] = None contract_end_date: Optional[datetime] = None class FormPhaseResponse(BaseModel): model_config = ConfigDict(from_attributes=True) budget_form_id: int sheet: str phase_code: str role: str column_keys: list[str] opens_at: datetime closes_at: datetime class FormPhaseCreate(BaseModel): sheet: str phase_code: str role: FormPhaseRole column_keys: list[str] opens_at: datetime closes_at: datetime class FormPhaseUpdate(BaseModel): role: FormPhaseRole | None = None column_keys: list[str] | None = None opens_at: datetime | None = None closes_at: datetime | None = None class AuditLogBase(BaseModel): """Базовая схема записи аудита (единый формат вывода как у auditlog).""" entity: str = Field(..., description="Тип сущности") entity_id: Optional[int] = Field(None, description="ID сущности") action: str = Field(..., description="Действие") payload_json: Optional[Dict[str, Any]] = None class AuditLogInDB(AuditLogBase): """Схема записи аудита в базе данных.""" model_config = ConfigDict(from_attributes=True) id: int = Field(..., description="ID записи") user_id: Optional[int] = Field(None, description="ID пользователя") at: datetime = Field(..., description="Дата/время события") class AuditLog(AuditLogInDB): """Схема записи аудита для ответа API.""" user: Optional[User] = None class AuditLogListResponse(ResponseBase): """Схема всех записей аудита для ответа API.""" result: List[AuditLog] = Field(..., description="Вывод записей аудита") class AuditLogQueryParams(BaseModel): """Query-параметры для фильтрации журнала аудита.""" page: int = Field(1, ge=1, description="Номер страницы") limit: int = Field( 20, ge=1, le=100, description="Количество записей на странице" ) user_id: Optional[int] = Field(None, description="ID пользователя") org_unit_id: Optional[int] = Field(None, description="ID ССП") task_id: Optional[int] = Field(None, description="ID задачи") form_id: Optional[int] = Field(None, description="ID формы") event_type: Optional[str] = Field(None, description="Тип события") date_from: Optional[datetime] = Field(None, description="Дата начала (ISO 8601)") date_to: Optional[datetime] = Field(None, description="Дата окончания (ISO 8601)")