Last active
April 17, 2019 15:14
-
-
Save delijati/0718258ca80008609a8fbc676c5192da to your computer and use it in GitHub Desktop.
dump all models when load sqlalchemy
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import os | |
import json | |
import hashlib | |
import datetime | |
from database import db | |
from sqlalchemy import event | |
def literal_sql(context): | |
statement = context.statement | |
session = context.session | |
try: | |
dialect = session.get_bind().dialect | |
compiled = statement.compile(dialect=dialect, | |
compile_kwargs={'literal_binds': True}) | |
return str(compiled).replace("\n", "") | |
except Exception: | |
return str(statement).replace("\n", "") | |
def dump(data, fp): | |
def default(o): | |
if isinstance(o, (datetime.date, datetime.datetime)): | |
return o.isoformat() | |
return json.dump( | |
data, | |
fp, | |
sort_keys=True, | |
indent=1, | |
default=default | |
) | |
@event.listens_for(db.Model, "load", propagate=True) | |
def _record_loaded_instances(target, context): | |
# from sqlalchemy.ext.serializer import loads, dumps | |
obj = {c.name: getattr(target, c.name) for c in target.__table__.columns} | |
literal = literal_sql(context) | |
perm = {} | |
if os.path.exists("dump.json"): | |
with open("dump.json", "r") as f: | |
perm = json.load(f) | |
key = hashlib.md5(literal.encode('utf-8')).hexdigest() | |
if key not in perm: | |
class_path = "%s.%s" % (target.__class__.__module__, | |
target.__class__.__name__) | |
perm[key] = {"sql": literal, "obj": obj, | |
"class": class_path} | |
with open("dump.json", "w") as f: | |
dump(perm, f) | |
print(literal) | |
print(obj) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment