247 lines
9.3 KiB
Python
247 lines
9.3 KiB
Python
from datetime import date
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import func, select, text, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
|
|
from src.db.models import Vsp
|
|
from src.domain.schemas import VSPCreate
|
|
|
|
|
|
class VSPRepository:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def get(
|
|
self,
|
|
vsp_id: int,
|
|
is_active: bool | None = None,
|
|
load_org_unit: bool = False,
|
|
with_deleted: bool = False,
|
|
) -> Optional[Vsp]:
|
|
queries = [Vsp.id == vsp_id, Vsp.is_deleted.is_(with_deleted)]
|
|
if is_active is not None:
|
|
queries.append(Vsp.is_active == is_active)
|
|
if load_org_unit:
|
|
return (
|
|
(await self.db.execute(
|
|
select(Vsp).where(*queries).options(joinedload(Vsp.org_unit)).limit(1))
|
|
).scalars().first()
|
|
)
|
|
else:
|
|
return (
|
|
(await self.db.execute(select(Vsp).where(*queries).limit(1))).scalars().first()
|
|
)
|
|
|
|
async def get_list(
|
|
self,
|
|
registration_number: str | None = None,
|
|
address: str | None = None,
|
|
ssp_ids: list[int] | None = None,
|
|
open_date_start: date | None = None,
|
|
open_date_end: date | None = None,
|
|
placement_type: str | None = None,
|
|
staff_count_min: int | None = None,
|
|
staff_count_max: int | None = None,
|
|
is_active: bool | None = True,
|
|
close_date_is_null: bool | None = None,
|
|
load_org_unit: bool = False,
|
|
with_count: bool = False,
|
|
) -> list[Vsp] | tuple[int, list[Vsp]]:
|
|
|
|
if with_count:
|
|
query = select(func.count().over().label("total_count"), Vsp)
|
|
else:
|
|
query = select(Vsp)
|
|
queries = [Vsp.is_deleted.is_(False)]
|
|
filters = [
|
|
("registration_number", registration_number),
|
|
("address", address),
|
|
("placement_type", placement_type),
|
|
("open_date_start", open_date_start),
|
|
("open_date_end", open_date_end),
|
|
("staff_count_min", staff_count_min),
|
|
("staff_count_max", staff_count_max),
|
|
("is_active", is_active),
|
|
]
|
|
for filter_name, filter_value in filters:
|
|
# print(filter_name, filter_value)
|
|
match filter_name, filter_value:
|
|
case ("registration_number", str(value)) if value:
|
|
queries.append(Vsp.reg_number.ilike(f"%{value}%"))
|
|
case ("address", str(value)) if value:
|
|
queries.append(Vsp.address.ilike(f"%{value}%"))
|
|
case ("placement_type", str(value)) if value:
|
|
queries.append(Vsp.placement_type.ilike(f"%{value}%"))
|
|
case ("open_date_start", value) if value is not None:
|
|
queries.append(Vsp.opened_at >= value)
|
|
case ("open_date_end", value) if value is not None:
|
|
queries.append(Vsp.opened_at <= value)
|
|
case ("staff_count_min", value) if value is not None:
|
|
queries.append(Vsp.staff_count >= value)
|
|
case ("staff_count_max", value) if value is not None:
|
|
queries.append(Vsp.staff_count <= value)
|
|
case ("is_active", value) if value is not None:
|
|
queries.append(Vsp.is_active == value)
|
|
case _:
|
|
continue
|
|
|
|
if close_date_is_null is True:
|
|
queries.append(Vsp.closed_at.is_(None))
|
|
elif close_date_is_null is False:
|
|
queries.append(Vsp.closed_at.is_not(None))
|
|
|
|
if ssp_ids is not None:
|
|
if not ssp_ids:
|
|
if with_count:
|
|
return 0, []
|
|
return []
|
|
queries.append(Vsp.branch_id.in_(ssp_ids))
|
|
|
|
query = query.where(*queries)
|
|
if load_org_unit:
|
|
query = query.options(joinedload(Vsp.org_unit))
|
|
query = query.order_by(Vsp.id)
|
|
|
|
if with_count:
|
|
result = (await self.db.execute(query)).all()
|
|
if result:
|
|
return result[0][0], [res[1] for res in result]
|
|
else:
|
|
count_query = select(func.count(Vsp.id))
|
|
if query.whereclause is not None:
|
|
count_query = count_query.where(
|
|
query.whereclause
|
|
)
|
|
return (await self.db.execute(count_query)).scalar(), []
|
|
else:
|
|
return (await self.db.execute(query)).scalars().all()
|
|
return (await self.db.execute(query.order_by(Vsp.id))).scalars().all()
|
|
|
|
async def create(self, payload: VSPCreate, created_by: int, load_org_unit: bool = False) -> Vsp:
|
|
data = payload.model_dump()
|
|
|
|
vsp_id = (
|
|
await self.db.execute(
|
|
text("""
|
|
SELECT v3.add_vsp(
|
|
:p_branch_id, :p_reg_number, :p_address, :p_format,
|
|
:p_opened_at, :p_placement_type, :p_staff_count,
|
|
:p_total_area, :p_closed_at, :p_is_active, :p_is_deleted,
|
|
:p_system_code, :p_created_by, :p_vsp_type, :p_notes,
|
|
:p_rent_contract_num, :p_rent_end_date
|
|
)
|
|
"""),
|
|
{
|
|
"p_branch_id": data.get("branch_id"),
|
|
"p_reg_number": data.get("reg_number"),
|
|
"p_address": data.get("address"),
|
|
"p_format": data.get("format"),
|
|
"p_opened_at": data.get("opened_at"),
|
|
"p_placement_type": data.get("placement_type"),
|
|
"p_staff_count": data.get("staff_count"),
|
|
"p_total_area": data.get("total_area"),
|
|
"p_closed_at": data.get("closed_at"),
|
|
"p_is_active": data.get("is_active", True),
|
|
"p_is_deleted": data.get("is_deleted", False),
|
|
"p_system_code": data.get("system_code"),
|
|
"p_created_by": created_by,
|
|
"p_vsp_type": data.get("vsp_type"),
|
|
"p_notes": data.get("notes"),
|
|
"p_rent_contract_num": data.get("rent_contract_num"),
|
|
"p_rent_end_date": data.get("rent_end_date"),
|
|
}
|
|
)
|
|
).scalar()
|
|
|
|
await self.db.flush()
|
|
|
|
if load_org_unit:
|
|
return (
|
|
await self.db.execute(
|
|
select(Vsp).where(Vsp.id == vsp_id).options(joinedload(Vsp.org_unit)).limit(1)
|
|
)
|
|
).scalars().first()
|
|
else:
|
|
return (
|
|
await self.db.execute(
|
|
select(Vsp).where(Vsp.id == vsp_id).limit(1)
|
|
)
|
|
).scalars().first()
|
|
|
|
async def update(self, vsp: Vsp, data: dict, updated_by: int, load_org_unit: bool = False) -> Vsp:
|
|
FIELD_TO_PARAM = {
|
|
"system_code": "p_system_code",
|
|
"vsp_type": "p_vsp_type",
|
|
"reg_number": "p_reg_number",
|
|
"address": "p_address",
|
|
"opened_at": "p_opened_at",
|
|
"closed_at": "p_closed_at",
|
|
"format": "p_format",
|
|
"placement_type": "p_placement_type",
|
|
"staff_count": "p_staff_count",
|
|
"notes": "p_notes",
|
|
"total_area": "p_total_area",
|
|
"rent_contract_num": "p_rent_contract_num",
|
|
"rent_end_date": "p_rent_end_date",
|
|
"branch_id": "p_branch_id",
|
|
"is_active": "p_is_active",
|
|
"is_deleted": "p_is_deleted",
|
|
}
|
|
|
|
params = {"p_vsp_id": vsp.id, "p_updated_by": updated_by}
|
|
for field, param in FIELD_TO_PARAM.items():
|
|
params[param] = data.get(field)
|
|
|
|
await self.db.execute(
|
|
text("""
|
|
SELECT v3.upd_vsp(
|
|
:p_vsp_id, :p_branch_id, :p_reg_number, :p_address,
|
|
:p_format, :p_opened_at, :p_placement_type, :p_staff_count,
|
|
:p_total_area, :p_closed_at, :p_is_active, :p_is_deleted,
|
|
:p_system_code, :p_updated_by, :p_vsp_type, :p_notes,
|
|
:p_rent_contract_num, :p_rent_end_date
|
|
)
|
|
"""),
|
|
params,
|
|
)
|
|
|
|
await self.db.flush()
|
|
await self.db.refresh(vsp)
|
|
|
|
if load_org_unit:
|
|
return (
|
|
await self.db.execute(
|
|
select(Vsp).where(Vsp.id == vsp.id).options(joinedload(Vsp.org_unit)).limit(1)
|
|
)
|
|
).scalars().first()
|
|
else:
|
|
return vsp
|
|
|
|
async def deactivate_by_close_date(self, as_of: date) -> int:
|
|
result = await self.db.execute(
|
|
update(Vsp)
|
|
.where(
|
|
Vsp.is_active.is_(True),
|
|
Vsp.closed_at.is_not(None),
|
|
Vsp.closed_at <= as_of,
|
|
)
|
|
.values(is_active=False, updated_at=func.now())
|
|
)
|
|
await self.db.commit()
|
|
return result.rowcount or 0
|
|
|
|
async def logical_delete(self, vsp_id: int, updated_by: int) -> bool:
|
|
result = (
|
|
await self.db.execute(
|
|
text("SELECT v3.del_vsp(:p_vsp_id, :p_deleted_by)"),
|
|
{"p_vsp_id": vsp_id, "p_deleted_by": updated_by},
|
|
)
|
|
).scalar()
|
|
await self.db.flush()
|
|
return result is not None
|
|
|
|
|