450 lines
11 KiB
Python
450 lines
11 KiB
Python
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"] = data.org_units
|
||
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
|
||
|
||
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)")
|