app.backbone.database.db_service
1from contextlib import contextmanager 2from typing import Iterable 3from sqlalchemy import UUID 4from sqlalchemy.orm import Session 5from app.backbone.database.db import get_db 6from sqlalchemy.sql import and_ 7 8class DbService: 9 def __init__(self): 10 pass # No necesita estar vinculado a un modelo especĂfico 11 12 def get_by_id(self, db: Session, model, id: int): 13 return db.query(model).filter(model.Id == id).first() 14 15 def get_by_filter(self, db: Session, model, **filters): 16 return db.query(model).filter(and_(*[getattr(model, key) == value for key, value in filters.items()])).first() 17 18 def get_many_by_filter(self, db: Session, model, **filters): 19 return db.query(model).filter(and_(*[getattr(model, key) == value for key, value in filters.items()])).all() 20 21 def delete_many_by_filter(self, db: Session, model, **filters): 22 return db.query(model).filter(and_(*[getattr(model, key) == value for key, value in filters.items()])).delete() 23 24 def get_all(self, db: Session, model): 25 return db.query(model).all() 26 27 def create(self, db: Session, obj_in): 28 db.add(obj_in) 29 return obj_in 30 31 def create_all(self, db, iterable: Iterable[object]) -> None: 32 db.add_all(iterable) 33 34 return None 35 36 def update(self, db: Session, model, obj_in): 37 db_obj = db.query(model).filter(model.Id == obj_in.Id).first() 38 db.merge(obj_in) 39 return db_obj 40 41 def delete(self, db: Session, model, id: UUID): 42 obj = db.query(model).filter(model.Id == id).first() 43 if obj: 44 db.delete(obj) 45 return obj 46 47 def delete_all(self, db: Session, model): 48 db.query(model).delete() 49 50 def save(self, db: Session): 51 try: 52 db.commit() 53 except Exception as e: 54 db.rollback() 55 raise e 56 57 @contextmanager 58 def get_database(self): 59 with get_db() as db: 60 try: 61 db.expire_on_commit = False 62 yield db 63 db.commit() 64 except Exception as e: 65 db.rollback() 66 raise e
class
DbService:
9class DbService: 10 def __init__(self): 11 pass # No necesita estar vinculado a un modelo especĂfico 12 13 def get_by_id(self, db: Session, model, id: int): 14 return db.query(model).filter(model.Id == id).first() 15 16 def get_by_filter(self, db: Session, model, **filters): 17 return db.query(model).filter(and_(*[getattr(model, key) == value for key, value in filters.items()])).first() 18 19 def get_many_by_filter(self, db: Session, model, **filters): 20 return db.query(model).filter(and_(*[getattr(model, key) == value for key, value in filters.items()])).all() 21 22 def delete_many_by_filter(self, db: Session, model, **filters): 23 return db.query(model).filter(and_(*[getattr(model, key) == value for key, value in filters.items()])).delete() 24 25 def get_all(self, db: Session, model): 26 return db.query(model).all() 27 28 def create(self, db: Session, obj_in): 29 db.add(obj_in) 30 return obj_in 31 32 def create_all(self, db, iterable: Iterable[object]) -> None: 33 db.add_all(iterable) 34 35 return None 36 37 def update(self, db: Session, model, obj_in): 38 db_obj = db.query(model).filter(model.Id == obj_in.Id).first() 39 db.merge(obj_in) 40 return db_obj 41 42 def delete(self, db: Session, model, id: UUID): 43 obj = db.query(model).filter(model.Id == id).first() 44 if obj: 45 db.delete(obj) 46 return obj 47 48 def delete_all(self, db: Session, model): 49 db.query(model).delete() 50 51 def save(self, db: Session): 52 try: 53 db.commit() 54 except Exception as e: 55 db.rollback() 56 raise e 57 58 @contextmanager 59 def get_database(self): 60 with get_db() as db: 61 try: 62 db.expire_on_commit = False 63 yield db 64 db.commit() 65 except Exception as e: 66 db.rollback() 67 raise e