148 lines
4.7 KiB
Python
148 lines
4.7 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
|
|
from app.models.journal import EntityType
|
|
from app.crud.crud_journal import journal
|
|
|
|
|
|
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, user: str = "system", user_ip: Optional[str] = None) -> Arrival:
|
|
from datetime import datetime as dt
|
|
|
|
update_data = obj_in.dict(exclude_unset=True)
|
|
changes = []
|
|
|
|
for field, value in update_data.items():
|
|
old_value = getattr(db_obj, field)
|
|
|
|
# Normalize datetime values for comparison (ignore timezone differences)
|
|
if isinstance(old_value, dt) and isinstance(value, dt):
|
|
# Compare only the date and time, ignoring timezone
|
|
old_normalized = old_value.replace(tzinfo=None) if old_value.tzinfo else old_value
|
|
new_normalized = value.replace(tzinfo=None) if value.tzinfo else value
|
|
if old_normalized == new_normalized:
|
|
continue # Skip if datetimes are the same
|
|
|
|
if old_value != value:
|
|
changes.append(f"{field} changed from '{old_value}' to '{value}'")
|
|
setattr(db_obj, field, value)
|
|
|
|
if changes:
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
|
|
# Log changes in journal
|
|
for change in changes:
|
|
journal.log_change(
|
|
db,
|
|
EntityType.ARRIVAL,
|
|
db_obj.id,
|
|
change,
|
|
user,
|
|
user_ip
|
|
)
|
|
|
|
return db_obj
|
|
|
|
def update_status(
|
|
self,
|
|
db: Session,
|
|
arrival_id: int,
|
|
status: ArrivalStatus,
|
|
timestamp: Optional[datetime] = None,
|
|
user: str = "system",
|
|
user_ip: Optional[str] = None
|
|
) -> Optional[Arrival]:
|
|
db_obj = self.get(db, arrival_id)
|
|
if not db_obj:
|
|
return None
|
|
|
|
old_status = db_obj.status
|
|
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)
|
|
|
|
# Log status change in journal
|
|
journal.log_change(
|
|
db,
|
|
EntityType.ARRIVAL,
|
|
arrival_id,
|
|
f"Status changed from {old_status.value} to {status.value}",
|
|
user,
|
|
user_ip
|
|
)
|
|
|
|
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()
|