137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
"""
|
|
Capa de servicio para lógica de negocio de localización
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .dto import LocationCreateDTO, LocationResponseDTO, LocationUpdateDTO
|
|
from .models import Location
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LocationService:
|
|
"""Servicio para gestión de localización"""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
@staticmethod
|
|
def get_all(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 50,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
) -> Tuple[List[Location], int]:
|
|
"""Get all locations with pagination"""
|
|
query = db.query(Location)
|
|
|
|
if filters:
|
|
if filters.get("code"):
|
|
query = query.filter(
|
|
Location.code.ilike(f"%{filters['code']}%"))
|
|
if filters.get("description"):
|
|
query = query.filter(
|
|
Location.description.ilike(f"%{filters['description']}%")
|
|
)
|
|
|
|
total = query.count()
|
|
locations = query.offset(skip).limit(limit).all()
|
|
|
|
return locations, total
|
|
|
|
@staticmethod
|
|
def get_by_id(db: Session, location_id: int) -> Optional[Location]:
|
|
"""Get location by ID"""
|
|
return db.query(Location).filter(Location.id == location_id).first()
|
|
|
|
@staticmethod
|
|
def get_by_code(db: Session, code: str) -> Optional[Location]:
|
|
"""Get location by code"""
|
|
return db.query(Location).filter(Location.code == code).first()
|
|
|
|
@staticmethod
|
|
def create(db: Session, location_data: LocationCreateDTO) -> Location:
|
|
"""Create a new location"""
|
|
try:
|
|
db_location = Location(
|
|
**location_data.model_dump(exclude_unset=True))
|
|
|
|
db.add(db_location)
|
|
db.commit()
|
|
db.refresh(db_location)
|
|
|
|
return db_location
|
|
|
|
except IntegrityError as e:
|
|
db.rollback()
|
|
logger.error(f"IntegrityError creating location: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Location code already exists",
|
|
)
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error creating location: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500, detail="Error creating location")
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session, location_id: int, location_data: LocationUpdateDTO
|
|
) -> Optional[Location]:
|
|
"""Update a location"""
|
|
try:
|
|
db_location = db.query(Location).filter(
|
|
Location.id == location_id).first()
|
|
|
|
if not db_location:
|
|
return None
|
|
|
|
for key, value in location_data.model_dump(exclude_unset=True).items():
|
|
setattr(db_location, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_location)
|
|
|
|
return db_location
|
|
|
|
except IntegrityError as e:
|
|
db.rollback()
|
|
logger.error(f"IntegrityError updating location: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Error updating location",
|
|
)
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error updating location: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500, detail="Error updating location")
|
|
|
|
@staticmethod
|
|
def delete(db: Session, location_id: int) -> bool:
|
|
"""Delete a location"""
|
|
try:
|
|
db_location = db.query(Location).filter(
|
|
Location.id == location_id).first()
|
|
|
|
if not db_location:
|
|
return False
|
|
|
|
db.delete(db_location)
|
|
db.commit()
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error deleting location: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500, detail="Error deleting location")
|