Last active
September 15, 2016 07:28
-
-
Save haruair/e673c2a131c7db230ffc21dec8ccd040 to your computer and use it in GitHub Desktop.
SQLAlchemy dirty convert using Normalizable
This file contains 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
from datetime import datetime | |
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey | |
from sqlalchemy.ext.hybrid import hybrid_property | |
from sqlalchemy.orm import relationship, backref | |
from database import Base | |
from util import Normalizable | |
class Address(Base, Normalizable): | |
__tablename__ = "addresses" | |
__excludes__ = ["password"] | |
id = Column(Integer, primary_key=True) | |
email_address = Column(String, nullable=False) | |
created_at = Column(DateTime) | |
user_id = Column(Integer, ForeignKey("users.id")) | |
user = relationship("User", backref=backref("addresses", order_by=id)) | |
@hybrid_property | |
def hello(self): | |
return 'email is %s' % self.email_address | |
def __init__(self, email_address): | |
self.email_address = email_address | |
self.created_at = datetime.now() | |
def __repr__(self): | |
return "<Address('%s')>" % (self.email_address) |
This file contains 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
from sqlalchemy import create_engine | |
from sqlalchemy.orm import sessionmaker | |
from database import Base | |
from user import User | |
from address import Address | |
engine = create_engine('sqlite:///tmp.db') | |
Base.metadata.create_all(engine) | |
Session = sessionmaker(bind=engine) | |
db_session = Session() | |
me = User('Edward', 'full name', '1234') | |
db_session.add(me) | |
db_session.commit() | |
me.addresses = [Address(email_address='edward@example.com'), | |
Address(email_address='haruair@example.com')] | |
db_session.add(me) | |
db_session.commit() | |
print(me.vars()) |
This file contains 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
from sqlalchemy import create_engine | |
from sqlalchemy.ext.declarative import declarative_base | |
from sqlalchemy.orm import sessionmaker | |
Base = declarative_base() |
This file contains 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
from datetime import datetime | |
from sqlalchemy import Column, Integer, String, DateTime | |
from sqlalchemy.ext.hybrid import hybrid_property | |
from database import Base | |
from util import Normalizable | |
class User(Base, Normalizable): | |
__tablename__ = "users" | |
__includes__ = ["hello", "addresses"] | |
__excludes__ = ["password"] | |
id = Column(Integer, primary_key=True) | |
name = Column(String) | |
fullname = Column(String) | |
password = Column(String) | |
created_at = Column(DateTime) | |
@hybrid_property | |
def hello(self): | |
return "my name is %s" % self.name | |
def __init__(self, name, fullname, password): | |
self.name = name | |
self.fullname = fullname | |
self.password = password | |
self.created_at = datetime.now() | |
def __repr__(self): | |
return "<User('%s', '%s', '%s')>" % (self.name, self.fullname, self.password) |
This file contains 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
from datetime import datetime | |
from sqlalchemy.orm.collections import InstrumentedList | |
from sqlalchemy import inspect | |
class Normalizable: | |
def field_normalize(self, x): | |
if isinstance(x, datetime): | |
return x.isoformat() | |
if isinstance(x, InstrumentedList): | |
return [item.vars() for item in x | |
if isinstance(item, Normalizable)] | |
else: | |
return x | |
def vars(self, includes=[], excludes=[], includes_only=None): | |
if hasattr(self, "__includes__"): | |
includes = includes + self.__includes__ | |
if hasattr(self, "__includes_only__"): | |
if includes_only is None: | |
includes_only = [] | |
includes_only = includes_only + self.__includes_only__ | |
if hasattr(self, "__excludes__"): | |
excludes = excludes + self.__excludes__ | |
includes_only = set(includes_only) | |
includes = set(includes).union(includes_only) | |
excludes = set(excludes) - includes | |
keys = {c.key for c in inspect(self).mapper.column_attrs} | |
keys = keys.union(includes) - excludes | |
if includes_only is not None: | |
keys = keys.intersection(includes_only) | |
fields = { key: self.field_normalize(getattr(self, key)) | |
for key in keys } | |
return fields |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment