49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
from sqlmodel import Field, SQLModel, create_engine
|
|
from sqlalchemy.ext.asyncio.engine import create_async_engine
|
|
from sqlalchemy.orm.session import sessionmaker
|
|
from sqlalchemy.ext.asyncio.session import AsyncSession
|
|
|
|
|
|
class DB:
|
|
# instance = None
|
|
config = None
|
|
engine = None
|
|
async_engine = None
|
|
session_maker = None
|
|
|
|
def __new__(cls):
|
|
if not hasattr(cls, 'instance'):
|
|
cls.instance = super(DB, cls).__new__(cls)
|
|
return cls.instance
|
|
|
|
def __init__(self):
|
|
from db import config
|
|
self.config = config
|
|
sqlite_url = self.config['DATABASE_URI']
|
|
|
|
self.engine = create_engine(sqlite_url, echo=True)
|
|
|
|
SQLModel.metadata.create_all(self.engine)
|
|
|
|
def get_engine(self):
|
|
return self.engine
|
|
|
|
def create_async_engine(self):
|
|
self.async_engine = create_async_engine(self.config['DATABASE_URI'], furure=True)
|
|
self.session_maker = sessionmaker(self.async_engine, expire_on_commit=False, class_=AsyncSession)
|
|
|
|
|
|
class DBException(Exception):
|
|
def __init__(self, *args):
|
|
self.is_db = True
|
|
if args:
|
|
self.msg = args[0]
|
|
else:
|
|
self.msg = None
|
|
|
|
def __str__(self):
|
|
if self.msg:
|
|
return "DB error, {0}".format(self.msg)
|
|
else:
|
|
return "DB error"
|