144 lines
4.2 KiB
Python
144 lines
4.2 KiB
Python
import logging
|
|
from typing import List, Optional, Dict, Any, Tuple
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .models import ConceptManifestation
|
|
from .dtos import (
|
|
ConceptManifestationCreateDTO,
|
|
ConceptManifestationUpdateDTO,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConceptManifestationService:
|
|
@staticmethod
|
|
def get_all(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
) -> Tuple[List[ConceptManifestation], int]:
|
|
query = db.query(ConceptManifestation).filter(
|
|
ConceptManifestation.tenant_id == tenant_id,
|
|
ConceptManifestation.company_id == company_id,
|
|
)
|
|
|
|
if filters:
|
|
if filters.get("value_manifestation_id"):
|
|
query = query.filter(
|
|
ConceptManifestation.value_manifestation_id == filters["value_manifestation_id"]
|
|
)
|
|
|
|
total = query.count()
|
|
items = query.offset(skip).limit(limit).all()
|
|
|
|
return items, total
|
|
|
|
@staticmethod
|
|
def get_by_id(
|
|
db: Session,
|
|
value_manifestation_id: int,
|
|
line_number: int,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> Optional[ConceptManifestation]:
|
|
return (
|
|
db.query(ConceptManifestation)
|
|
.filter(
|
|
ConceptManifestation.value_manifestation_id == value_manifestation_id,
|
|
ConceptManifestation.line_number == line_number,
|
|
ConceptManifestation.tenant_id == tenant_id,
|
|
ConceptManifestation.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
def create(
|
|
db: Session,
|
|
data: ConceptManifestationCreateDTO,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> ConceptManifestation:
|
|
try:
|
|
db_item = ConceptManifestation(
|
|
**data.model_dump(exclude_unset=True),
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
db.add(db_item)
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
return db_item
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(f"Error creating concept manifestation: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500, detail="Error creating concept manifestation"
|
|
)
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session,
|
|
value_manifestation_id: int,
|
|
line_number: int,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
data: ConceptManifestationUpdateDTO,
|
|
) -> Optional[ConceptManifestation]:
|
|
item = ConceptManifestationService.get_by_id(
|
|
db, value_manifestation_id, line_number, tenant_id, company_id
|
|
)
|
|
if not item:
|
|
return None
|
|
|
|
try:
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(item, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(
|
|
f"Error updating concept manifestation {value_manifestation_id}-{line_number}: {str(e)}"
|
|
)
|
|
raise HTTPException(
|
|
status_code=500, detail="Error updating concept manifestation"
|
|
)
|
|
|
|
@staticmethod
|
|
def delete(
|
|
db: Session,
|
|
value_manifestation_id: int,
|
|
line_number: int,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> bool:
|
|
item = ConceptManifestationService.get_by_id(
|
|
db, value_manifestation_id, line_number, tenant_id, company_id
|
|
)
|
|
if not item:
|
|
return False
|
|
|
|
try:
|
|
db.delete(item)
|
|
db.commit()
|
|
return True
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.error(
|
|
f"Error deleting concept manifestation {value_manifestation_id}-{line_number}: {str(e)}"
|
|
)
|
|
raise HTTPException(
|
|
status_code=500, detail="Error deleting concept manifestation"
|
|
)
|