49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
from fastapi import HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .dto import ItemCreate, ItemUpdate
|
|
from .models import Item
|
|
|
|
|
|
def get_items(db: Session, tenant_id: int, company_id: int) -> list[Item]:
|
|
return (
|
|
db.query(Item)
|
|
.filter(Item.tenant_id == tenant_id, Item.company_id == company_id, Item.deleted_at.is_(None))
|
|
.all()
|
|
)
|
|
|
|
|
|
def get_item(db: Session, item_id: int, tenant_id: int, company_id: int) -> Item:
|
|
item = (
|
|
db.query(Item)
|
|
.filter(Item.id == item_id, Item.tenant_id == tenant_id, Item.company_id == company_id, Item.deleted_at.is_(None))
|
|
.first()
|
|
)
|
|
if not item:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item no encontrado")
|
|
return item
|
|
|
|
|
|
def create_item(db: Session, payload: ItemCreate, tenant_id: int, company_id: int) -> Item:
|
|
item = Item(**payload.model_dump(), tenant_id=tenant_id, company_id=company_id)
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
def update_item(db: Session, item_id: int, payload: ItemUpdate, tenant_id: int, company_id: int) -> Item:
|
|
item = get_item(db, item_id, tenant_id, company_id)
|
|
for field, value in payload.model_dump(exclude_unset=True).items():
|
|
setattr(item, field, value)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
def delete_item(db: Session, item_id: int, tenant_id: int, company_id: int) -> None:
|
|
item = get_item(db, item_id, tenant_id, company_id)
|
|
from datetime import datetime, timezone
|
|
item.deleted_at = datetime.now(timezone.utc)
|
|
db.commit()
|