136 lines
4.4 KiB
Python
136 lines
4.4 KiB
Python
from typing import List, Optional, Dict, Any
|
|
from uuid import UUID
|
|
from sqlalchemy.orm import Session
|
|
from models.logistics import Logistics
|
|
from schemas.logistics import LogisticsCreate, LogisticsUpdate, LogisticsSchema
|
|
|
|
def get_all_logistics(db: Session) -> List[LogisticsSchema]:
|
|
"""
|
|
Retrieves all logistics from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[LogisticsSchema]: A list of all logistics schemas.
|
|
"""
|
|
logistics = db.query(Logistics).all()
|
|
return [LogisticsSchema.from_orm(logistics_obj) for logistics_obj in logistics]
|
|
|
|
def get_logistics_by_id(db: Session, logistics_id: UUID) -> Optional[LogisticsSchema]:
|
|
"""
|
|
Retrieves a single logistics by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
logistics_id (UUID): The ID of the logistics to retrieve.
|
|
|
|
Returns:
|
|
Optional[LogisticsSchema]: The logistics schema if found, otherwise None.
|
|
"""
|
|
logistics = db.query(Logistics).filter(Logistics.id == logistics_id).first()
|
|
if logistics:
|
|
return LogisticsSchema.from_orm(logistics)
|
|
else:
|
|
return None
|
|
|
|
def create_logistics(db: Session, logistics_data: LogisticsCreate) -> LogisticsSchema:
|
|
"""
|
|
Creates a new logistics in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
logistics_data (LogisticsCreate): The data for the logistics to create.
|
|
|
|
Returns:
|
|
LogisticsSchema: The newly created logistics schema.
|
|
"""
|
|
db_logistics = Logistics(**logistics_data.dict())
|
|
db.add(db_logistics)
|
|
db.commit()
|
|
db.refresh(db_logistics)
|
|
return LogisticsSchema.from_orm(db_logistics)
|
|
|
|
def update_logistics(db: Session, logistics_id: UUID, logistics_data: LogisticsUpdate) -> Optional[LogisticsSchema]:
|
|
"""
|
|
Updates an existing logistics in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
logistics_id (UUID): The ID of the logistics to update.
|
|
logistics_data (LogisticsUpdate): The updated data for the logistics.
|
|
|
|
Returns:
|
|
Optional[LogisticsSchema]: The updated logistics schema if found, otherwise None.
|
|
"""
|
|
logistics = db.query(Logistics).filter(Logistics.id == logistics_id).first()
|
|
if logistics:
|
|
update_data = logistics_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(logistics, key, value)
|
|
db.commit()
|
|
db.refresh(logistics)
|
|
return LogisticsSchema.from_orm(logistics)
|
|
else:
|
|
return None
|
|
|
|
def delete_logistics(db: Session, logistics_id: UUID) -> bool:
|
|
"""
|
|
Deletes a logistics from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
logistics_id (UUID): The ID of the logistics to delete.
|
|
|
|
Returns:
|
|
bool: True if the logistics was deleted, False otherwise.
|
|
"""
|
|
logistics = db.query(Logistics).filter(Logistics.id == logistics_id).first()
|
|
if logistics:
|
|
db.delete(logistics)
|
|
db.commit()
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def validate_logistics_data(logistics_data: Dict[str, Any]) -> bool:
|
|
"""
|
|
Validates input logistics data dictionary.
|
|
|
|
Args:
|
|
logistics_data (Dict[str, Any]): The input logistics data to validate.
|
|
|
|
Returns:
|
|
bool: True if the data is valid, False otherwise.
|
|
"""
|
|
if not logistics_data:
|
|
return False
|
|
if "name" not in logistics_data or not isinstance(logistics_data["name"], str) or len(logistics_data["name"]) < 3:
|
|
return False
|
|
# Add more validation rules as needed
|
|
return True
|
|
|
|
def generate_logistics_report(logistics_data: List[LogisticsSchema]) -> Dict[str, Any]:
|
|
"""
|
|
Generates a report based on the provided logistics data.
|
|
|
|
Args:
|
|
logistics_data (List[LogisticsSchema]): The logistics data to generate the report from.
|
|
|
|
Returns:
|
|
Dict[str, Any]: The generated logistics report.
|
|
"""
|
|
report = {
|
|
"total_logistics": len(logistics_data),
|
|
"logistics_by_creation_date": {}
|
|
}
|
|
for logistics in logistics_data:
|
|
creation_date = logistics.created_at.date()
|
|
if creation_date not in report["logistics_by_creation_date"]:
|
|
report["logistics_by_creation_date"][creation_date] = []
|
|
report["logistics_by_creation_date"][creation_date].append({
|
|
"id": logistics.id,
|
|
"name": logistics.name,
|
|
"description": logistics.description
|
|
})
|
|
return report |