Cheat Sheet For Busy Coders: SQLAlchemy #SQL #mysql

csfbc_sqlalchemy.py

#! /usr/bin/python

# 02/27/2015, Kenial
# Just simple scripts for instant use of SQLAlchemy.

import sqlalchemy as sa
sa.__version__  # '0.9.4' for me

# Sample DB connection string
# engine = sa.create_engine('postgresql://scott:tiger@localhost/mydatabase')
# engine = sa.create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')
# engine = sa.create_engine('postgresql+pg8000://scott:tiger@localhost/mydatabase')
# engine = sa.create_engine('mysql://scott:tiger@localhost/foo?charset=utf8')
# engine = sa.create_engine('mysql+mysqldb://scott:tiger@localhost/?charset=utf8')
# engine = sa.create_engine('mysql+pymysql://scott:tiger@localhost/?charset=utf8')
# engine = sa.create_engine('mysql+mysqlconnector://scott:tiger@localhost/foo')
# engine = sa.create_engine('mysql+oursql://scott:tiger@localhost/foo')


##################################################
# 'traditional' database definition / manipulation by SQL
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///sqlite.db', echo=True)
conn = engine.connect()

# you got OperationalError if SQL is incorrect or table exists
result = conn.execute("""
    CREATE TABLE user (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        first_name TEXT,
        last_name TEXT,
        job TEXT
    );
""")

# result.rowcount will be 1
result = conn.execute("""
    INSERT INTO user (first_name, last_name, job) values ('Kenial', 'Lee', 'SE');
""")
# due to stupid sqlite3_excute API ... it can just run one statement per one call
result = conn.execute("""
    INSERT INTO user (first_name, last_name, job) values ('Jon', 'Skeet', 'Author');
""")

result = conn.execute("""
    SELECT * FROM user;
""")

print result.rowcount   # should be 2, but returns -1. due to stupid sqlite3 API ...

# Iterate them all
row = result.fetchone()
while row:
    print row
    row = result.fetchone()
# Otherwise, you got [(1, u'Kenial', u'Lee', u'SE'), (2, u'Jon', u'Skeet', u'Author')]
print result.fetchall()

# result.rowcount will be 2
result = conn.execute("""
    DELETE FROM user;
""")
conn.close()

######################################
# Semi - statement mapping ways
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///sqlite.db', echo=True)
metadata = sa.MetaData(bind=engine)
users = sa.Table(
    'user', 
    metadata,
    sa.Column('id', sa.Integer, primary_key=True, autoincrement=True),
    sa.Column('first_name', sa.String),
    sa.Column('last_name', sa.String),
    sa.Column('job', sa.String),
)
metadata.create_all()     # create table - it's safe to run many times

# insert(), select() methods return Insert, Select class object, which 
# represents SQL statement template and behaves like Command, Template 
# class of traditional DB library. 
users.insert(values={
    "first_name": "Kenial",
    "last_name": "Lee",
    "job": "SE",
}).execute()
result = users.select().execute()

# you got [(1, u'Kenial', u'Lee', u'SE')]
print result.fetchall()

# Delete - should refer to c, which stands for 'column'
users.delete(users.c.first_name == "Kenial").execute()

######################################
# ORM ways here
import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy.ext.declarative import declarative_base

engine = sa.create_engine('sqlite:///sqlite.db', echo=True)
metadata = sa.MetaData(bind=engine)
db_session = orm.scoped_session(orm.sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
))

class User(object):
    id = sa.Column(sa.Integer, primary_key=True)
    first_name = sa.Column(sa.String)
    last_name = sa.Column(sa.String)
    job = sa.Column(sa.String)
    #
    def __init__(self, first_name=None, last_name=None, job=None):
        self.first_name = first_name
        self.last_name = last_name
        self.job = job
    #
    def __repr__(self):
        return '<User %r>' % (self.name)
    #
    query = sa.Table(
        'user', 
        metadata,
        sa.Column('id', sa.Integer, primary_key=True, autoincrement=True),
        sa.Column('first_name', sa.String),
        sa.Column('last_name', sa.String),
        sa.Column('job', sa.String),
    )


def init_db():
    metadata.create_all(bind=engine)


# set up db and schema
init_db()
orm.mapper(User, User.query)

u = User("Kenial", "Lee", "SE")
db_session.add(User("Kenial", "Lee", "SE"))
db_session.add_all([
    User("Kenial", "Lee", "SE"),
    User("Jon", "Skeet", "Author"),
])
db_session.commit()

result = User.query.select(User.id == 1).execute()
# you got [(1, u'Kenial', u'Lee', u'SE')]
print result.fetchall()

# looks like ORM now!
result = User.query.delete(User.id == 1).execute()



##############################################################
# "DB first" code

from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy.orm.query import Query
from sqlalchemy import desc
from sqlalchemy import text

# engine, suppose it has two tables 'user' and 'address' set up
engine = create_engine('mysql+pymysql://scott:tiger@localhost/dbname?charset=utf8')

Base = automap_base()
# reflect the tables
Base.prepare(engine, reflect=True)

# mapped classes are now created with names by default
# matching that of the table name.
SomeObj = Base.classes.some_table

session = Session(engine)

some_obj_list = session.query(SomeObj)[:10]
some_obj_list = session.query(SomeObj.column_name).order_by(desc(SomeObj.id))[:10]


session.query(SomeObj).filter(SomeObj.id.in_([10, 20]))
session.query(SomeObj.id).filter_by(id=15)

# rudimentary relationships are produced
session.add(SomeObj(id=15, foreign_obj=ForeignObj(name="foo")))
session.commit()


# to check UPDATE, use session.dirty
# to check INSERT, use session.new

# bulk INSERT
session.add_all([

])


# collection-based relationships are by default named
# "<classname>_collection"
some_obj = some_obj_list[0]
print (u1.foreign_obj_collection)