Skip to content

Instantly share code, notes, and snippets.

@delijati
Last active April 17, 2019 15:14
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save delijati/0718258ca80008609a8fbc676c5192da to your computer and use it in GitHub Desktop.
Save delijati/0718258ca80008609a8fbc676c5192da to your computer and use it in GitHub Desktop.
dump all models when load sqlalchemy
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