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
def get_by_id(self, db: sqlalchemy.orm.session.Session, model, id: int):
13    def get_by_id(self, db: Session, model, id: int):
14        return db.query(model).filter(model.Id == id).first()
def get_by_filter(self, db: sqlalchemy.orm.session.Session, model, **filters):
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()
def get_many_by_filter(self, db: sqlalchemy.orm.session.Session, model, **filters):
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()
def delete_many_by_filter(self, db: sqlalchemy.orm.session.Session, model, **filters):
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()
def get_all(self, db: sqlalchemy.orm.session.Session, model):
25    def get_all(self, db: Session, model):
26        return db.query(model).all()
def create(self, db: sqlalchemy.orm.session.Session, obj_in):
28    def create(self, db: Session, obj_in):
29        db.add(obj_in)
30        return obj_in
def create_all(self, db, iterable: Iterable[object]) -> None:
32    def create_all(self, db, iterable: Iterable[object]) -> None:
33        db.add_all(iterable)
34        
35        return None
def update(self, db: sqlalchemy.orm.session.Session, model, obj_in):
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
def delete( self, db: sqlalchemy.orm.session.Session, model, id: sqlalchemy.sql.sqltypes.UUID):
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
def delete_all(self, db: sqlalchemy.orm.session.Session, model):
48    def delete_all(self, db: Session, model):
49        db.query(model).delete()
def save(self, db: sqlalchemy.orm.session.Session):
51    def save(self, db: Session):
52        try:
53            db.commit()
54        except Exception as e:
55            db.rollback()
56            raise e
@contextmanager
def get_database(self):
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