63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
from sqlalchemy.orm import Session
|
|
from typing import Optional, List, Tuple
|
|
from uuid import UUID
|
|
from src.models.cpt_code_model import CPTCode
|
|
from src.validation.cpt_code_schemas import CPTCodeCreate, CPTCodeUpdate
|
|
|
|
class CPTCodeCRUD:
|
|
"""CRUD operations for CPTCode"""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def get_all(
|
|
self,
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) -> Tuple[List[CPTCode], int]:
|
|
"""Get all cptcodes with pagination"""
|
|
query = self.db.query(CPTCode)
|
|
total = query.count()
|
|
items = query.order_by(CPTCode.created_at.desc()).offset(skip).limit(limit).all()
|
|
return items, total
|
|
|
|
def get_by_id(self, cpt_code_id: UUID) -> Optional[CPTCode]:
|
|
"""Get cptcode by ID"""
|
|
return self.db.query(CPTCode).filter(CPTCode.id == cpt_code_id).first()
|
|
|
|
def create(self, cpt_code_in: CPTCodeCreate) -> CPTCode:
|
|
"""Create a new cptcode"""
|
|
db_cpt_code = CPTCode(**cpt_code_in.model_dump())
|
|
self.db.add(db_cpt_code)
|
|
self.db.commit()
|
|
self.db.refresh(db_cpt_code)
|
|
return db_cpt_code
|
|
|
|
def update(
|
|
self,
|
|
cpt_code_id: UUID,
|
|
cpt_code_in: CPTCodeUpdate
|
|
) -> Optional[CPTCode]:
|
|
"""Update an existing cptcode"""
|
|
db_cpt_code = self.get_by_id(cpt_code_id)
|
|
if not db_cpt_code:
|
|
return None
|
|
|
|
update_data = cpt_code_in.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_cpt_code, field, value)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(db_cpt_code)
|
|
return db_cpt_code
|
|
|
|
def delete(self, cpt_code_id: UUID) -> bool:
|
|
"""Delete a cptcode"""
|
|
db_cpt_code = self.get_by_id(cpt_code_id)
|
|
if not db_cpt_code:
|
|
return False
|
|
|
|
self.db.delete(db_cpt_code)
|
|
self.db.commit()
|
|
return True
|