105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import and_, or_, func, desc
|
|
from datetime import date, datetime
|
|
from app.models.arrival import Arrival, ArrivalStatus
|
|
from app.schemas.arrival import ArrivalCreate, ArrivalUpdate, ArrivalStatusUpdate
|
|
|
|
|
|
class CRUDArrival:
|
|
def get(self, db: Session, arrival_id: int) -> Optional[Arrival]:
|
|
return db.query(Arrival).filter(Arrival.id == arrival_id).first()
|
|
|
|
def get_multi(
|
|
self,
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
status: Optional[ArrivalStatus] = None,
|
|
date_from: Optional[date] = None,
|
|
date_to: Optional[date] = None
|
|
) -> List[Arrival]:
|
|
query = db.query(Arrival)
|
|
|
|
if status:
|
|
query = query.filter(Arrival.status == status)
|
|
|
|
if date_from:
|
|
query = query.filter(func.date(Arrival.created_dt) >= date_from)
|
|
|
|
if date_to:
|
|
query = query.filter(func.date(Arrival.created_dt) <= date_to)
|
|
|
|
return query.order_by(desc(Arrival.created_dt)).offset(skip).limit(limit).all()
|
|
|
|
def get_arrivals_today(self, db: Session) -> List[Arrival]:
|
|
"""Get today's arrivals (booked in or landed)"""
|
|
today = date.today()
|
|
return db.query(Arrival).filter(
|
|
and_(
|
|
func.date(Arrival.created_dt) == today,
|
|
or_(
|
|
Arrival.status == ArrivalStatus.BOOKED_IN,
|
|
Arrival.status == ArrivalStatus.LANDED
|
|
)
|
|
)
|
|
).order_by(Arrival.created_dt).all()
|
|
|
|
def create(self, db: Session, obj_in: ArrivalCreate, created_by: str) -> Arrival:
|
|
db_obj = Arrival(
|
|
**obj_in.dict(),
|
|
created_by=created_by,
|
|
status=ArrivalStatus.BOOKED_IN
|
|
)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update(self, db: Session, db_obj: Arrival, obj_in: ArrivalUpdate) -> Arrival:
|
|
update_data = obj_in.dict(exclude_unset=True)
|
|
|
|
for field, value in update_data.items():
|
|
if value is not None:
|
|
setattr(db_obj, field, value)
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update_status(
|
|
self,
|
|
db: Session,
|
|
arrival_id: int,
|
|
status: ArrivalStatus,
|
|
timestamp: Optional[datetime] = None
|
|
) -> Optional[Arrival]:
|
|
db_obj = self.get(db, arrival_id)
|
|
if not db_obj:
|
|
return None
|
|
|
|
db_obj.status = status
|
|
|
|
if status == ArrivalStatus.LANDED and timestamp:
|
|
db_obj.landed_dt = timestamp
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def cancel(self, db: Session, arrival_id: int) -> Optional[Arrival]:
|
|
db_obj = self.get(db, arrival_id)
|
|
if not db_obj:
|
|
return None
|
|
|
|
db_obj.status = ArrivalStatus.CANCELLED
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
|
|
arrival = CRUDArrival()
|