1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
import click
import os
import glob
import time
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from flask_sqlalchemy import SQLAlchemy
from app.settings import app_cfg
@click.command('migrate_to_mysql')
@click.pass_context
def cli(ctx):
"""
- Create connections to both databases
- For each table, for each row, insert from one to the other
"""
mysql_session, mysql_base = make_mysql_base()
sqlite_session, sqlite_base = make_sqlite3_base()
mysql_classes = make_classes(mysql_base)
sqlite_classes = make_classes(sqlite_base)
for mysql_class, sqlite_class in zip(mysql_classes, sqlite_classes):
sqlite_objs = sqlite_session.query(sqlite_class).order_by(sqlite_class.id).all()
for sqlite_obj in sqlite_objs:
mysql_obj = mysql_class()
for column in sqlite_class.__table__.columns:
table_name, column_name = str(column).split(".")
# print(f"{table_name} => {column_name}")
# if column_name != 'id':
setattr(mysql_obj, column_name, getattr(sqlite_obj, column_name))
mysql_session.add(mysql_obj)
mysql_session.commit()
def make_mysql_base():
"""Make a Mysql connection"""
connection_url = "mysql+pymysql://{}:{}@{}/{}?charset=utf8mb4".format(
os.getenv("DB_USER"),
os.getenv("DB_PASS"),
os.getenv("DB_HOST"),
os.getenv("DB_NAME")
)
return make_base(connection_url)
def make_sqlite3_base():
"""Make a SQLite3 connection"""
connection_url = "sqlite:///{}".format(os.path.join(app_cfg.DIR_DATABASE, 'swimmer.sqlite3'))
return make_base(connection_url)
def make_base(connection_url):
"""Make a connection base from a connection URL"""
engine = create_engine(connection_url, encoding="utf-8", pool_recycle=3600)
Session = sessionmaker(bind=engine)
Base = declarative_base()
Base.metadata.bind = engine
db = SQLAlchemy()
return Session(), Base
def make_classes(Base):
"""Make classes from a base"""
from sqlalchemy import create_engine, Table, Column, Text, String, Integer, \
Boolean, Float, DateTime, JSON, ForeignKey
from sqlalchemy_utc import UtcDateTime, utcnow
class Upload(Base):
"""Table for storing references to various media"""
__tablename__ = 'upload'
id = Column(Integer, primary_key=True)
graph_id = Column(Integer, ForeignKey('graph.id'), nullable=True)
sha256 = Column(String(256), nullable=False)
fn = Column(String(256), nullable=False)
ext = Column(String(4, convert_unicode=True), nullable=False)
tag = Column(String(64, convert_unicode=True), nullable=True)
username = Column(String(16, convert_unicode=True), nullable=False)
created_at = Column(UtcDateTime(), default=utcnow())
class Tile(Base):
"""Table for storing references to tiles"""
__tablename__ = 'tile'
id = Column(Integer, primary_key=True)
graph_id = Column(Integer, ForeignKey('graph.id'), nullable=True)
page_id = Column(Integer, ForeignKey('page.id'), nullable=True)
target_page_id = Column(Integer, ForeignKey('page.id'), nullable=True)
type = Column(String(16, convert_unicode=True), nullable=False)
sort_order = Column(Integer, default=0)
settings = Column(JSON, default={}, nullable=True)
created_at = Column(UtcDateTime(), default=utcnow())
updated_at = Column(UtcDateTime(), onupdate=utcnow())
class Page(Base):
"""Table for storing references to pages"""
__tablename__ = 'page'
id = Column(Integer, primary_key=True)
graph_id = Column(Integer, ForeignKey('graph.id'), nullable=True)
path = Column(String(64, convert_unicode=True), nullable=False)
title = Column(String(64, convert_unicode=True), nullable=False)
username = Column(String(32, convert_unicode=True), nullable=False)
description = Column(Text(convert_unicode=True), nullable=False)
settings = Column(JSON, default={}, nullable=True)
created_at = Column(UtcDateTime(), default=utcnow())
updated_at = Column(UtcDateTime(), onupdate=utcnow())
class Graph(Base):
"""Table for storing references to graphs"""
__tablename__ = 'graph'
id = Column(Integer, primary_key=True)
home_page_id = Column(Integer, nullable=True)
path = Column(String(64, convert_unicode=True), nullable=False)
title = Column(String(64, convert_unicode=True), nullable=False)
username = Column(String(32, convert_unicode=True), nullable=False)
description = Column(Text(convert_unicode=True), nullable=False)
settings = Column(JSON, default={}, nullable=True)
created_at = Column(UtcDateTime(), default=utcnow())
updated_at = Column(UtcDateTime(), onupdate=utcnow())
return [ Graph, Page, Tile, Upload ]
|
