72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
from datetime import datetime
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.orm import scoped_session
|
|
from model.base import Base
|
|
from sqlalchemy import MetaData
|
|
from sqlalchemy import Table
|
|
from sqlalchemy import Index
|
|
from sqlalchemy import text
|
|
|
|
|
|
class Database:
|
|
def __init__(self, database_url="sqlite:///finance.db"):
|
|
self.engine = create_engine(database_url)
|
|
self.session = scoped_session(sessionmaker(bind=self.engine))
|
|
|
|
|
|
def init_app(self, app):
|
|
self.engine = create_engine((app.config["DATABASE_URI"]))
|
|
self.session = scoped_session(sessionmaker(bind=self.engine))
|
|
|
|
|
|
# flask request handlers.
|
|
@app.after_request
|
|
def after_request(response):
|
|
self.session.remove()
|
|
return response
|
|
|
|
|
|
def create_tables(self):
|
|
Base.metadata.create_all(self.engine)
|
|
|
|
|
|
def drop_tables(self):
|
|
Base.metadata.drop_all(self.engine)
|
|
|
|
|
|
def drop_table(self, table):
|
|
meta = MetaData()
|
|
meta.reflect(bind=self.engine)
|
|
if table in meta.tables:
|
|
table = Table(table, meta, autoload_with=self.engine)
|
|
table.drop(self.engine, checkfirst=True)
|
|
|
|
|
|
def copy_table_structure(self, table, copy_table):
|
|
|
|
metadata = MetaData()
|
|
existing_table = Table(table, metadata, autoload_with=self.engine)
|
|
|
|
new_table = Table(
|
|
copy_table, metadata,
|
|
*(c.copy() for c in existing_table.columns),
|
|
extend_existing=True
|
|
)
|
|
|
|
new_table.create(self.engine, checkfirst=True)
|
|
|
|
for index in existing_table.indexes:
|
|
new_index_name = f"{index.name}_{copy_table}"
|
|
new_index = Index(
|
|
new_index_name,
|
|
*[new_table.c[col.name] for col in index.columns],
|
|
unique=index.unique
|
|
)
|
|
new_index.create(self.engine)
|
|
|
|
|
|
def rename_table(self, old_name, new_name):
|
|
with self.engine.connect() as conn:
|
|
conn.execute(text(f"DROP TABLE IF EXISTS {new_name}"))
|
|
conn.execute(text(f"ALTER TABLE {old_name} RENAME TO {new_name}")) |