monitoring/config/db_session.py

37 lines
700 B
Python

# _*_ coding: utf-8 _*_
# @Time :2022/5/29 11:26
# @Email :508737091@qq.com
# @Author :qiangyanwen
# @File :db_session.py.py
from typing import Generator
from config.database import SessionLocal
from sqlalchemy.orm import Session
def get_db() -> Generator:
db = SessionLocal()
try:
yield db
finally:
db.close()
def db_add(db: Session, model: object):
db.add(model)
db.commit()
db.refresh(model)
def db_save(db: Session, model: object):
db.commit()
db.flush()
db.refresh(model)
def db_delete(db: Session, model: object):
db.delete(model)
db.commit()
db.flush()
__all__ = ["get_db", "db_add", "db_save","db_delete"]