41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
"""当前用户关联驾押人员所挂靠车辆的车牌集合。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional, Set
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.pl_models import PlVehicle, PlVehiclePersonnel
|
|
from app.services.pl_driver_plan_service import resolve_personnel_for_user
|
|
|
|
|
|
def license_plates_for_user_personnel(
|
|
db: Session, tenant_no: str, user_id: str
|
|
) -> Set[str]:
|
|
personnel = resolve_personnel_for_user(db, tenant_no=tenant_no, user_id=user_id)
|
|
if not personnel:
|
|
return set()
|
|
rows = (
|
|
db.query(PlVehicle.license_plate)
|
|
.join(
|
|
PlVehiclePersonnel,
|
|
PlVehiclePersonnel.vehicle_id == PlVehicle.id,
|
|
)
|
|
.filter(
|
|
PlVehiclePersonnel.personnel_id == personnel.id,
|
|
PlVehiclePersonnel.tenant_no == tenant_no,
|
|
PlVehiclePersonnel.del_flag == 0,
|
|
PlVehicle.tenant_no == tenant_no,
|
|
PlVehicle.del_flag == 0,
|
|
)
|
|
.all()
|
|
)
|
|
return {str(r[0]).strip() for r in rows if r and r[0] and str(r[0]).strip()}
|
|
|
|
|
|
def resolve_personnel_id_for_user(
|
|
db: Session, tenant_no: str, user_id: str
|
|
) -> Optional[str]:
|
|
p = resolve_personnel_for_user(db, tenant_no=tenant_no, user_id=user_id)
|
|
return p.id if p else None
|