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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
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')
@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, 'animism.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
# from app.sql.common import db, Base, Session
class Episode(Base):
"""Table for storing episodes and their metadata"""
__tablename__ = 'episode'
id = Column(Integer, primary_key=True)
episode_number = Column(Integer)
title = Column(String(256, convert_unicode=True), nullable=False)
release_date = Column(String(256, convert_unicode=True))
is_live = Column(Boolean, default=False)
settings = Column(JSON, default={}, nullable=True)
class Annotation(Base):
"""Table for storing references to annotations"""
__tablename__ = 'annotation'
id = Column(Integer, primary_key=True)
type = Column(String(16, convert_unicode=True), nullable=False)
paragraph_id = Column(Integer, nullable=True)
start_ts = Column(Float, nullable=False)
end_ts = Column(Float, nullable=True)
text = Column(Text(convert_unicode=True), nullable=True)
settings = Column(JSON, default={}, nullable=True)
class Media(Base):
"""Table for storing references to media"""
__tablename__ = 'media'
id = Column(Integer, primary_key=True)
type = Column(String(16, convert_unicode=True), nullable=False)
tag = Column(String(64, convert_unicode=True), nullable=True)
url = Column(String(256, convert_unicode=True), nullable=True)
title = Column(String(256, convert_unicode=True), nullable=True)
author = Column(String(256, convert_unicode=True), nullable=True)
pre_title = Column(String(256, convert_unicode=True), nullable=True)
post_title = Column(String(256, convert_unicode=True), nullable=True)
translated_title = Column(String(256, convert_unicode=True), nullable=True)
date = Column(String(256, convert_unicode=True), nullable=True)
source = Column(String(256, convert_unicode=True), nullable=True)
medium = Column(String(64, convert_unicode=True), nullable=True)
description = Column(Text(convert_unicode=True), nullable=True)
start_ts = Column(Float, nullable=True)
settings = Column(JSON, default={}, nullable=True)
class Paragraph(Base):
"""Table for storing paragraphs, which contain annotations"""
__tablename__ = 'paragraph'
id = Column(Integer, primary_key=True)
type = Column(String(16, convert_unicode=True), nullable=False)
start_ts = Column(Float, nullable=False)
end_ts = Column(Float, nullable=True)
settings = Column(JSON, default={}, nullable=True)
class Upload(Base):
"""Table for storing references to various media"""
__tablename__ = 'upload'
id = Column(Integer, primary_key=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 User(Base):
"""Table for storing the user list"""
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
username = Column(String(256, convert_unicode=True), nullable=False)
password = Column(String(256, convert_unicode=True), nullable=False)
is_admin = Column(Boolean, default=False)
settings = Column(JSON, default={}, nullable=True)
class Venue(Base):
"""Table for storing the venue list"""
__tablename__ = 'venue'
id = Column(Integer, primary_key=True)
title = Column(String(256, convert_unicode=True), nullable=False)
date = Column(String(256, convert_unicode=True), nullable=False)
settings = Column(JSON, default={}, nullable=True)
return [ Episode, Paragraph, Annotation, Media, Upload, User, Venue ]
|
