SQLAlchemy MySQL Cheat Sheet

csfbc_sqlalchemy.md

Easy readable md file for sqlalchemy.py based on kenial/csfbc_sqlalchemy.py [https://gist.github.com/kenial/db76f51f4d05e6f0bb67]

Cheat Sheet For Busy Coders: SQLAlchemy #SQL #mysql

#! /usr/bin/python
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

    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

"_collection"

some_obj = some_obj_list[0]
print (u1.foreign_obj_collection)