latosa/db.py
2024-03-03 11:28:05 +01:00

60 lines
2.2 KiB
Python

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]