46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
|
|
|
|
class SqlalchemyConnect:
|
|
def __init__(self, Base: declarative_base, host="127.0.0.1", user="", password="", db=""):
|
|
self.Base = Base
|
|
self.host = host
|
|
self.user = user
|
|
self.password = password
|
|
self.db = db
|
|
self.engine = self.init_engine()
|
|
|
|
def init_engine(self):
|
|
engine = create_engine(
|
|
f"mysql+pymysql://{self.user}:{self.password}@{self.host}/{self.db}?charset=utf8mb4")
|
|
return engine
|
|
|
|
def get_db(self) -> sessionmaker:
|
|
try:
|
|
session = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)
|
|
db = session()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
def get_db_commit(self) -> sessionmaker:
|
|
try:
|
|
session = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)
|
|
db = session()
|
|
yield db
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
def get_db_i(self) -> Session:
|
|
engine = create_engine(
|
|
f"mysql+pymysql://{self.user}:{self.password}@{self.host}/{self.db}?charset=utf8mb4")
|
|
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
db = session()
|
|
return db
|
|
|
|
def init_database(self):
|
|
self.Base.metadata.create_all(bind=self.engine)
|