Skip to content

Instantly share code, notes, and snippets.

@aleiphoenix aleiphoenix/app.py
Created Apr 2, 2015

Embed
What would you like to do?
native SQLAlchemy ORM usage...
# -*- coding: utf-8 -*-
import json
import logging
from sqlalchemy import create_engine
from sqlalchemy import types
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.schema import MetaData, Table
from sqlalchemy import Column, Integer
from sqlalchemy.orm import sessionmaker, scoped_session
logging.basicConfig(
format='[%(asctime)s %(levelname)-5s] %(message)s',
level=logging.DEBUG
)
logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
logging.debug(logging.getLogger('sqlalchemy.engine').handlers)
# engine = create_engine("mysql+mysqlconnector://root@127.0.0.1:3306/adv")
engine = create_engine("sqlite:///app.db", echo=True)
Base = declarative_base(bind=engine)
logging.debug(Base.metadata)
class JSON(types.UserDefinedType):
def get_col_spec(self):
return 'JSON'
def bind_processor(self, dialect):
def process(value):
return json.dumps(value)
return process
def result_process(self, dialect, coltype):
def process(value):
return json.loads(value)
return process
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, autoincrement=True)
contact = Column(JSON)
Base.metadata.create_all()
SessionFactory = sessionmaker(bind=engine)
Session = scoped_session(SessionFactory)
session = Session()
user = User()
user.contact = {
'phone': 'xxx',
'address': 'yyy'
}
# session.add(user)
# session.flush()
# session.commit()
user = session.query(User).first()
logging.debug(user.contact)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.