You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
61 lines
2.2 KiB
61 lines
2.2 KiB
from os import mkdir
|
|
from os.path import isdir
|
|
|
|
from sqlalchemy import (BigInteger, Boolean, Column, Integer, MetaData, String,
|
|
Table, create_engine, inspect)
|
|
from sqlalchemy.engine.reflection import Inspector
|
|
from sqlalchemy.sql import text
|
|
from sqlalchemy_utils import create_database, database_exists
|
|
|
|
|
|
class DB:
|
|
|
|
def __init__(self, db_url=None):
|
|
if db_url is None:
|
|
db_path = '/app/db'
|
|
if not isdir(db_path):
|
|
mkdir(db_path)
|
|
db_url = f'sqlite:///{db_path}/latosa.db'
|
|
db_engine = db_url.split('://')[0]
|
|
if not database_exists(db_url):
|
|
create_database(db_url)
|
|
self.engine = create_engine(db_url, echo=True)
|
|
self.inspector: Inspector = inspect(self.engine)
|
|
self.tables = {}
|
|
tables = [
|
|
Table(
|
|
'auth', MetaData(),
|
|
Column('id',
|
|
BigInteger().with_variant(Integer, db_engine),
|
|
primary_key=True), Column('client_id', String),
|
|
Column('client_secret', String)),
|
|
Table(
|
|
'users', MetaData(),
|
|
Column('id',
|
|
BigInteger().with_variant(Integer, db_engine),
|
|
primary_key=True), Column('uid', String),
|
|
Column('email', String), Column('groups', String),
|
|
Column('password_hash', String), Column('salt', String),
|
|
Column('is_active', Boolean), Column('is_admin', Boolean),
|
|
Column('is_anonymous', Boolean),
|
|
Column('display_name', String))
|
|
]
|
|
if not database_exists(db_url):
|
|
create_database(db_url)
|
|
for table in tables:
|
|
self.tables[table.name] = table
|
|
if not self.inspector.has_table(table.name):
|
|
table.metadata.create_all(self.engine)
|
|
|
|
def connect(self):
|
|
return self.engine.connect()
|
|
|
|
def execute(self, s: str, bind_params: dict):
|
|
conn = self.connect()
|
|
res = conn.execute(text(s), bind_params)
|
|
conn.commit()
|
|
return res
|
|
|
|
def get_table(self, table_name):
|
|
return self.tables[table_name]
|