39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.customer import Customer
|
|
from app.schemas.customer import CustomerCreate, CustomerUpdate
|
|
|
|
|
|
class CustomerService:
|
|
def get(self, db: Session, customer_id: int) -> Optional[Customer]:
|
|
return db.query(Customer).filter(Customer.id == customer_id).first()
|
|
|
|
def get_multi_by_user(self, db: Session, user_id: int, skip: int = 0, limit: int = 100) -> List[Customer]:
|
|
return db.query(Customer).filter(Customer.user_id == user_id).offset(skip).limit(limit).all()
|
|
|
|
def create(self, db: Session, obj_in: CustomerCreate, user_id: int) -> Customer:
|
|
db_obj = Customer(**obj_in.dict(), user_id=user_id)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update(self, db: Session, db_obj: Customer, obj_in: CustomerUpdate) -> Customer:
|
|
update_data = obj_in.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_obj, field, value)
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def remove(self, db: Session, customer_id: int) -> Customer:
|
|
obj = db.query(Customer).get(customer_id)
|
|
db.delete(obj)
|
|
db.commit()
|
|
return obj
|
|
|
|
|
|
customer_service = CustomerService() |