acazares
962f43fe62
- Updated SealService to support tenant and company filtering with pagination and enhanced CRUD methods. - Refactored TrailerService to include tenant and company support, added filtering capabilities, and improved CRUD methods. - Introduced TenantCRUDRoutes for Trailer and Transporter routes to streamline API endpoint creation and management. - Enhanced TransporterService with tenant and company filtering, pagination, and improved CRUD operations. - Added Customs Broker module with DTOs, models, services, and routes for managing customs broker data. - Implemented CRUD operations for Customs Broker, including personnel and VU management. - Improved data validation and descriptions in DTOs for better API documentation.
115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
from typing import Optional, Tuple, List, Dict, Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import dto, models
|
|
|
|
|
|
class PermissionRuleOctService:
|
|
"""Service for PermissionRuleOct CRUD operations with tenant support"""
|
|
|
|
@staticmethod
|
|
def get_all(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
skip: int = 0,
|
|
limit: int = 50,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
) -> Tuple[List[models.PermissionRuleOct], int]:
|
|
"""Get all permission rules OCT for a tenant/company with pagination"""
|
|
query = db.query(models.PermissionRuleOct).filter(
|
|
models.PermissionRuleOct.tenant_id == tenant_id,
|
|
models.PermissionRuleOct.company_id == company_id,
|
|
)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if filters.get("permission"):
|
|
query = query.filter(
|
|
models.PermissionRuleOct.permission.ilike(f"%{filters['permission']}%")
|
|
)
|
|
if filters.get("sector"):
|
|
query = query.filter(
|
|
models.PermissionRuleOct.sector == filters["sector"]
|
|
)
|
|
if filters.get("system"):
|
|
query = query.filter(
|
|
models.PermissionRuleOct.system == filters["system"]
|
|
)
|
|
|
|
total = query.count()
|
|
permissions = query.offset(skip).limit(limit).all()
|
|
|
|
return permissions, total
|
|
|
|
@staticmethod
|
|
def get_by_id(
|
|
db: Session, permission_id: int, tenant_id: int, company_id: int
|
|
) -> Optional[models.PermissionRuleOct]:
|
|
"""Get permission rule OCT by ID"""
|
|
return (
|
|
db.query(models.PermissionRuleOct)
|
|
.filter(
|
|
models.PermissionRuleOct.id == permission_id,
|
|
models.PermissionRuleOct.tenant_id == tenant_id,
|
|
models.PermissionRuleOct.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
def create(
|
|
db: Session,
|
|
permission_data: dto.PermissionRuleOctCreateDTO,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> models.PermissionRuleOct:
|
|
"""Create a new permission rule OCT"""
|
|
new_permission = models.PermissionRuleOct(
|
|
**permission_data.model_dump(), tenant_id=tenant_id, company_id=company_id
|
|
)
|
|
db.add(new_permission)
|
|
db.commit()
|
|
db.refresh(new_permission)
|
|
return new_permission
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session,
|
|
permission_id: int,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
permission_data: dto.PermissionRuleOctUpdateDTO,
|
|
) -> Optional[models.PermissionRuleOct]:
|
|
"""Update a permission rule OCT"""
|
|
permission = PermissionRuleOctService.get_by_id(
|
|
db, permission_id, tenant_id, company_id
|
|
)
|
|
if not permission:
|
|
return None
|
|
|
|
# Update fields
|
|
update_data = permission_data.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(permission, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(permission)
|
|
return permission
|
|
|
|
@staticmethod
|
|
def delete(
|
|
db: Session, permission_id: int, tenant_id: int, company_id: int
|
|
) -> bool:
|
|
"""Delete a permission rule OCT"""
|
|
permission = PermissionRuleOctService.get_by_id(
|
|
db, permission_id, tenant_id, company_id
|
|
)
|
|
if not permission:
|
|
return False
|
|
|
|
db.delete(permission)
|
|
db.commit()
|
|
return True
|