docs/notes/python-sqlalchemy.rst
.. meta::
:description lang=en: Collect useful snippets of SQLAlchemy
:keywords: Python, Python3, SQLAlchemy Cheat Sheet
==========
SQLAlchemy
==========
.. contents:: Table of Contents
:backlinks: none
Set a database URL
-------------------
.. code-block:: python
from sqlalchemy.engine.url import URL
postgres_db = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
print(URL(**postgres_db))
sqlite_db = {'drivername': 'sqlite', 'database': 'db.sqlite'}
print(URL(**sqlite_db))
output:
.. code-block:: bash
$ python sqlalchemy_url.py
postgres://postgres:postgres@192.168.99.100:5432
sqlite:///db.sqlite
Sqlalchemy Support DBAPI - PEP249
-----------------------------------
.. code-block:: python
from sqlalchemy import create_engine
db_uri = "sqlite:///db.sqlite"
engine = create_engine(db_uri)
# DBAPI - PEP249
# create table
engine.execute('CREATE TABLE "EX1" ('
'id INTEGER NOT NULL,'
'name VARCHAR, '
'PRIMARY KEY (id));')
# insert a raw
engine.execute('INSERT INTO "EX1" '
'(id, name) '
'VALUES (1,"raw1")')
# select *
result = engine.execute('SELECT * FROM '
'"EX1"')
for _r in result:
print(_r)
# delete *
engine.execute('DELETE from "EX1" where id=1;')
result = engine.execute('SELECT * FROM "EX1"')
print(result.fetchall())
Transaction and Connect Object
--------------------------------
.. code-block:: python
from sqlalchemy import create_engine
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
# Create connection
conn = engine.connect()
# Begin transaction
trans = conn.begin()
conn.execute('INSERT INTO "EX1" (name) '
'VALUES ("Hello")')
trans.commit()
# Close connection
conn.close()
Metadata - Generating Database Schema
--------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer, String
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
# Create a metadata instance
metadata = MetaData(engine)
# Declare a table
table = Table('Example',metadata,
Column('id',Integer, primary_key=True),
Column('name',String))
# Create all tables
metadata.create_all()
for _t in metadata.tables:
print("Table: ", _t)
Inspect - Get Database Information
------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import inspect
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
inspector = inspect(engine)
# Get table information
print(inspector.get_table_names())
# Get column information
print(inspector.get_columns('EX1'))
Reflection - Loading Table from Existing Database
---------------------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
# Create a MetaData instance
metadata = MetaData()
print(metadata.tables)
# reflect db schema to MetaData
metadata.reflect(bind=engine)
print(metadata.tables)
Print Create Table Statement with Indexes (SQL DDL)
----------------------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
def metadata_dump(sql, *multiparams, **params):
print(sql.compile(dialect=engine.dialect))
meta = MetaData()
example_table = Table('Example',meta,
Column('id', Integer, primary_key=True),
Column('name', String(10), index=True))
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri, strategy='mock', executor=metadata_dump)
meta.create_all(bind=engine, tables=[example_table])
output:
.. code-block:: sql
CREATE TABLE "Example" (
id INTEGER NOT NULL,
name VARCHAR(10),
PRIMARY KEY (id)
)
CREATE INDEX "ix_Example_name" ON "Example" (name)
Get Table from MetaData
------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
# Create MetaData instance
metadata = MetaData(engine).reflect()
print(metadata.tables)
# Get Table
ex_table = metadata.tables['Example']
print(ex_table)
Create all Tables Store in "MetaData"
--------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer, String
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
meta = MetaData(engine)
# Register t1, t2 to metadata
t1 = Table('EX1', meta,
Column('id',Integer, primary_key=True),
Column('name',String))
t2 = Table('EX2', meta,
Column('id',Integer, primary_key=True),
Column('val',Integer))
# Create all tables in meta
meta.create_all()
Create Specific Table
-----------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer, String
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
meta = MetaData(engine)
t1 = Table('Table_1', meta,
Column('id', Integer, primary_key=True),
Column('name',String))
t2 = Table('Table_2', meta,
Column('id', Integer, primary_key=True),
Column('val',Integer))
t1.create()
Create table with same columns
-------------------------------
.. code-block:: python
from sqlalchemy import (
create_engine,
inspect,
Column,
String,
Integer)
from sqlalchemy.ext.declarative import declarative_base
db_url = "sqlite://"
engine = create_engine(db_url)
Base = declarative_base()
class TemplateTable(object):
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
class DowntownAPeople(TemplateTable, Base):
__tablename__ = "downtown_a_people"
class DowntownBPeople(TemplateTable, Base):
__tablename__ = "downtown_b_people"
Base.metadata.create_all(bind=engine)
# check table exists
ins = inspect(engine)
for _t in ins.get_table_names():
print(_t)
Drop a Table
-------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import inspect
from sqlalchemy import Table
from sqlalchemy import Column, Integer, String
from sqlalchemy.engine.url import URL
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
m = MetaData()
table = Table('Test', m,
Column('id', Integer, primary_key=True),
Column('key', String, nullable=True),
Column('val', String))
table.create(engine)
inspector = inspect(engine)
print('Test' in inspector.get_table_names())
table.drop(engine)
inspector = inspect(engine)
print('Test' in inspector.get_table_names())
output:
.. code-block:: bash
$ python sqlalchemy_drop.py
$ True
$ False
Some Table Object Operation
----------------------------
.. code-block:: python
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer, String
meta = MetaData()
t = Table('ex_table', meta,
Column('id', Integer, primary_key=True),
Column('key', String),
Column('val', Integer))
# Get Table Name
print(t.name)
# Get Columns
print(t.columns.keys())
# Get Column
c = t.c.key
print(c.name)
# Or
c = t.columns.key
print(c.name)
# Get Table from Column
print(c.table)
SQL Expression Language
-------------------------
.. code-block:: python
# Think Column as "ColumnElement"
# Implement via overwrite special function
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer, String
from sqlalchemy import or_
meta = MetaData()
table = Table('example', meta,
Column('id', Integer, primary_key=True),
Column('l_name', String),
Column('f_name', String))
# sql expression binary object
print(repr(table.c.l_name == 'ed'))
# exhbit sql expression
print(str(table.c.l_name == 'ed'))
print(repr(table.c.f_name != 'ed'))
# comparison operator
print(repr(table.c.id > 3))
# or expression
print((table.c.id > 5) | (table.c.id < 2))
# Equal to
print(or_(table.c.id > 5, table.c.id < 2))
# compare to None produce IS NULL
print(table.c.l_name == None)
# Equal to
print(table.c.l_name.is_(None))
# + means "addition"
print(table.c.id + 5)
# or means "string concatenation"
print(table.c.l_name + "some name")
# in expression
print(table.c.l_name.in_(['a','b']))
insert() - Create an "INSERT" Statement
----------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
# create table
meta = MetaData(engine)
table = Table('user', meta,
Column('id', Integer, primary_key=True),
Column('l_name', String),
Column('f_name', String))
meta.create_all()
# insert data via insert() construct
ins = table.insert().values(
l_name='Hello',
f_name='World')
conn = engine.connect()
conn.execute(ins)
# insert multiple data
conn.execute(table.insert(),[
{'l_name':'Hi','f_name':'bob'},
{'l_name':'yo','f_name':'alice'}])
select() - Create a "SELECT" Statement
---------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import select
from sqlalchemy import or_
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
conn = engine.connect()
meta = MetaData(engine).reflect()
table = meta.tables['user']
# select * from 'user'
select_st = select([table]).where(
table.c.l_name == 'Hello')
res = conn.execute(select_st)
for _row in res:
print(_row)
# or equal to
select_st = table.select().where(
table.c.l_name == 'Hello')
res = conn.execute(select_st)
for _row in res:
print(_row)
# combine with "OR"
select_st = select([
table.c.l_name,
table.c.f_name]).where(or_(
table.c.l_name == 'Hello',
table.c.l_name == 'Hi'))
res = conn.execute(select_st)
for _row in res:
print(_row)
# combine with "ORDER_BY"
select_st = select([table]).where(or_(
table.c.l_name == 'Hello',
table.c.l_name == 'Hi')).order_by(table.c.f_name)
res = conn.execute(select_st)
for _row in res:
print(_row)
join() - Joined Two Tables via "JOIN" Statement
------------------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import select
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
meta = MetaData(engine).reflect()
email_t = Table('email_addr', meta,
Column('id', Integer, primary_key=True),
Column('email',String),
Column('name',String))
meta.create_all()
# get user table
user_t = meta.tables['user']
# insert
conn = engine.connect()
conn.execute(email_t.insert(),[
{'email':'ker@test','name':'Hi'},
{'email':'yo@test','name':'Hello'}])
# join statement
join_obj = user_t.join(email_t,
email_t.c.name == user_t.c.l_name)
# using select_from
sel_st = select(
[user_t.c.l_name, email_t.c.email]).select_from(join_obj)
res = conn.execute(sel_st)
for _row in res:
print(_row)
Fastest Bulk Insert in PostgreSQL via "COPY" Statement
-------------------------------------------------------
.. code-block:: python
# This method found here: https://gist.github.com/jsheedy/efa9a69926a754bebf0e9078fd085df6
import io
from datetime import date
from sqlalchemy.engine.url import URL
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import Date
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
# create table
meta = MetaData(engine)
table = Table('userinfo', meta,
Column('id', Integer, primary_key=True),
Column('first_name', String),
Column('age', Integer),
Column('birth_day', Date),
)
meta.create_all()
# file-like object (tsv format)
datafile = io.StringIO()
# generate rows
for i in range(100):
line = '\t'.join(
[
f'Name {i}', # first_name
str(18 + i), # age
str(date.today()), # birth_day
]
)
datafile.write(line + '\n')
# reset file to start
datafile.seek(0)
# bulk insert via `COPY` statement
conn = engine.raw_connection()
with conn.cursor() as cur:
# https://www.psycopg.org/docs/cursor.html#cursor.copy_from
cur.copy_from(
datafile,
table.name, # table name
sep='\t',
columns=('first_name', 'age', 'birth_day'),
)
conn.commit()
Bulk PostgreSQL Insert and Return Inserted IDs
-----------------------------------------------
.. code-block:: python
from sqlalchemy.engine.url import URL
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
# create table
meta = MetaData(engine)
table = Table('userinfo', meta,
Column('id', Integer, primary_key=True),
Column('first_name', String),
Column('age', Integer),
)
meta.create_all()
# generate rows
data = [{'first_name': f'Name {i}', 'age': 18+i} for i in range(10)]
stmt = table.insert().values(data).returning(table.c.id)
# converted into SQL:
# INSERT INTO userinfo (first_name, age) VALUES
# (%(first_name_m0)s, %(age_m0)s), (%(first_name_m1)s, %(age_m1)s),
# (%(first_name_m2)s, %(age_m2)s), (%(first_name_m3)s, %(age_m3)s),
# (%(first_name_m4)s, %(age_m4)s), (%(first_name_m5)s, %(age_m5)s),
# (%(first_name_m6)s, %(age_m6)s), (%(first_name_m7)s, %(age_m7)s),
# (%(first_name_m8)s, %(age_m8)s), (%(first_name_m9)s, %(age_m9)s)
# RETURNING userinfo.id
for rowid in engine.execute(stmt).fetchall():
print(rowid['id'])
output:
.. code-block:: bash
$ python sqlalchemy_bulk.py
1
2
3
4
5
6
7
8
9
10
Update Multiple Rows
---------------------
.. code-block:: python
from sqlalchemy.engine.url import URL
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.sql.expression import bindparam
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
# create table
meta = MetaData(engine)
table = Table('userinfo', meta,
Column('id', Integer, primary_key=True),
Column('first_name', String),
Column('birth_year', Integer),
)
meta.create_all()
# update data
data = [
{'_id': 1, 'first_name': 'Johnny', 'birth_year': 1975},
{'_id': 2, 'first_name': 'Jim', 'birth_year': 1973},
{'_id': 3, 'first_name': 'Kaley', 'birth_year': 1985},
{'_id': 4, 'first_name': 'Simon', 'birth_year': 1980},
{'_id': 5, 'first_name': 'Kunal', 'birth_year': 1981},
{'_id': 6, 'first_name': 'Mayim', 'birth_year': 1975},
{'_id': 7, 'first_name': 'Melissa', 'birth_year': 1980},
]
stmt = table.update().where(table.c.id == bindparam('_id')).\
values({
'first_name': bindparam('first_name'),
'birth_year': bindparam('birth_year'),
})
# conveted to SQL:
# UPDATE userinfo SET first_name=%(first_name)s, birth_year=%(birth_year)s WHERE userinfo.id = %(_id)s
engine.execute(stmt, data)
Delete Rows from Table
------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
conn = engine.connect()
meta = MetaData(engine).reflect()
user_t = meta.tables['user']
# select * from user_t
sel_st = user_t.select()
res = conn.execute(sel_st)
for _row in res:
print(_row)
# delete l_name == 'Hello'
del_st = user_t.delete().where(
user_t.c.l_name == 'Hello')
print('----- delete -----')
res = conn.execute(del_st)
# check rows has been delete
sel_st = user_t.select()
res = conn.execute(sel_st)
for _row in res:
print(_row)
Check Table Existing
----------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Column
from sqlalchemy import Integer, String
from sqlalchemy import inspect
from sqlalchemy.ext.declarative import declarative_base
Modal = declarative_base()
class Example(Modal):
__tablename__ = "ex_t"
id = Column(Integer, primary_key=True)
name = Column(String(20))
db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
Modal.metadata.create_all(engine)
# check register table exist to Modal
for _t in Modal.metadata.tables:
print(_t)
# check all table in database
meta = MetaData(engine).reflect()
for _t in meta.tables:
print(_t)
# check table names exists via inspect
ins = inspect(engine)
for _t in ins.get_table_names():
print(_t)
Create multiple tables at once
-------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import inspect
from sqlalchemy import Column, String, Integer
from sqlalchemy.engine.url import URL
db = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
url = URL(**db)
engine = create_engine(url)
metadata = MetaData()
metadata.reflect(bind=engine)
def create_table(name, metadata):
tables = metadata.tables.keys()
if name not in tables:
table = Table(name, metadata,
Column('id', Integer, primary_key=True),
Column('key', String),
Column('val', Integer))
table.create(engine)
tables = ['table1', 'table2', 'table3']
for _t in tables: create_table(_t, metadata)
inspector = inspect(engine)
print(inspector.get_table_names())
output:
.. code-block:: bash
$ python sqlalchemy_create.py
[u'table1', u'table2', u'table3']
Create tables with dynamic columns (Table)
--------------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String
from sqlalchemy import Table
from sqlalchemy import MetaData
from sqlalchemy import inspect
from sqlalchemy.engine.url import URL
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
def create_table(name, *cols):
meta = MetaData()
meta.reflect(bind=engine)
if name in meta.tables: return
table = Table(name, meta, *cols)
table.create(engine)
create_table('Table1',
Column('id', Integer, primary_key=True),
Column('name', String))
create_table('Table2',
Column('id', Integer, primary_key=True),
Column('key', String),
Column('val', String))
inspector = inspect(engine)
for _t in inspector.get_table_names():
print(_t)
output:
.. code-block:: bash
$ python sqlalchemy_dynamic.py
Table1
Table2
Object Relational add data
----------------------------
.. code-block:: python
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine.url import URL
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
Base = declarative_base()
class TestTable(Base):
__tablename__ = 'Test Table'
id = Column(Integer, primary_key=True)
key = Column(String, nullable=False)
val = Column(String)
date = Column(DateTime, default=datetime.utcnow)
# create tables
Base.metadata.create_all(bind=engine)
# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
data = {'a': 5566, 'b': 9527, 'c': 183}
try:
for _key, _val in data.items():
row = TestTable(key=_key, val=_val)
session.add(row)
session.commit()
except SQLAlchemyError as e:
print(e)
finally:
session.close()
Object Relational update data
------------------------------
.. code-block:: python
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine.url import URL
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
Base = declarative_base()
class TestTable(Base):
__tablename__ = 'Test Table'
id = Column(Integer, primary_key=True)
key = Column(String, nullable=False)
val = Column(String)
date = Column(DateTime, default=datetime.utcnow)
# create tables
Base.metadata.create_all(bind=engine)
# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
try:
# add row to database
row = TestTable(key="hello", val="world")
session.add(row)
session.commit()
# update row to database
row = session.query(TestTable).filter(
TestTable.key == 'hello').first()
print('original:', row.key, row.val)
row.key = "Hello"
row.val = "World"
session.commit()
# check update correct
row = session.query(TestTable).filter(
TestTable.key == 'Hello').first()
print('update:', row.key, row.val)
except SQLAlchemyError as e:
print(e)
finally:
session.close()
output:
.. code-block:: bash
$ python sqlalchemy_update.py
original: hello world
update: Hello World
Object Relational delete row
-----------------------------
.. code-block:: python
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine.url import URL
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
Base = declarative_base()
class TestTable(Base):
__tablename__ = 'Test Table'
id = Column(Integer, primary_key=True)
key = Column(String, nullable=False)
val = Column(String)
date = Column(DateTime, default=datetime.utcnow)
# create tables
Base.metadata.create_all(bind=engine)
# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
row = TestTable(key='hello', val='world')
session.add(row)
query = session.query(TestTable).filter(
TestTable.key=='hello')
print(query.first())
query.delete()
query = session.query(TestTable).filter(
TestTable.key=='hello')
print(query.all())
output:
.. code-block:: bash
$ python sqlalchemy_delete.py
<__main__.TestTable object at 0x104eb8f50>
[]
Object Relational relationship
-------------------------------
.. code-block:: python
from sqlalchemy import Column, String, Integer, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
name = Column(String)
addresses = relationship("Address", backref="user")
class Address(Base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True)
email = Column(String)
user_id = Column(Integer, ForeignKey('user.id'))
u1 = User()
a1 = Address()
print(u1.addresses)
print(a1.user)
u1.addresses.append(a1)
print(u1.addresses)
print(a1.user)
output:
.. code-block:: bash
$ python sqlalchemy_relationship.py
[]
None
[<__main__.Address object at 0x10c4edb50>]
<__main__.User object at 0x10c4ed810>
Object Relational self association
-----------------------------------
.. code-block:: python
import json
from sqlalchemy import (
Column,
Integer,
String,
ForeignKey,
Table)
from sqlalchemy.orm import (
sessionmaker,
relationship)
from sqlalchemy.ext.declarative import declarative_base
base = declarative_base()
association = Table("Association", base.metadata,
Column('left', Integer, ForeignKey('node.id'), primary_key=True),
Column('right', Integer, ForeignKey('node.id'), primary_key=True))
class Node(base):
__tablename__ = 'node'
id = Column(Integer, primary_key=True)
label = Column(String)
friends = relationship('Node',
secondary=association,
primaryjoin=id==association.c.left,
secondaryjoin=id==association.c.right,
backref='left')
def to_json(self):
return dict(id=self.id,
friends=[_.label for _ in self.friends])
nodes = [Node(label='node_{}'.format(_)) for _ in range(0, 3)]
nodes[0].friends.extend([nodes[1], nodes[2]])
nodes[1].friends.append(nodes[2])
print('----> right')
print(json.dumps([_.to_json() for _ in nodes], indent=2))
print('----> left')
print(json.dumps([_n.to_json() for _n in nodes[1].left], indent=2))
output:
.. code-block:: bash
----> right
[
{
"friends": [
"node_1",
"node_2"
],
"id": null
},
{
"friends": [
"node_2"
],
"id": null
},
{
"friends": [],
"id": null
}
]
----> left
[
{
"friends": [
"node_1",
"node_2"
],
"id": null
}
]
Object Relational basic query
------------------------------
.. code-block:: python
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy import Column, String, Integer, DateTime
from sqlalchemy import or_
from sqlalchemy import desc
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine.url import URL
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
Base = declarative_base()
class User(Base):
__tablename__ = 'User'
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
fullname = Column(String, nullable=False)
birth = Column(DateTime)
# create tables
engine = create_engine(URL(**db_url))
Base.metadata.create_all(bind=engine)
users = [
User(name='ed',
fullname='Ed Jones',
birth=datetime(1989,7,1)),
User(name='wendy',
fullname='Wendy Williams',
birth=datetime(1983,4,1)),
User(name='mary',
fullname='Mary Contrary',
birth=datetime(1990,1,30)),
User(name='fred',
fullname='Fred Flinstone',
birth=datetime(1977,3,12)),
User(name='justin',
fullname="Justin Bieber")]
# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
# add_all
session.add_all(users)
session.commit()
print("----> order_by(id):")
query = session.query(User).order_by(User.id)
for _row in query.all():
print(_row.name, _row.fullname, _row.birth)
print("\n----> order_by(desc(id)):")
query = session.query(User).order_by(desc(User.id))
for _row in query.all():
print(_row.name, _row.fullname, _row.birth)
print("\n----> order_by(date):")
query = session.query(User).order_by(User.birth)
for _row in query.all():
print(_row.name, _row.fullname, _row.birth)
print("\n----> EQUAL:")
query = session.query(User).filter(User.id == 2)
_row = query.first()
print(_row.name, _row.fullname, _row.birth)
print("\n----> NOT EQUAL:")
query = session.query(User).filter(User.id != 2)
for _row in query.all():
print(_row.name, _row.fullname, _row.birth)
print("\n----> IN:")
query = session.query(User).filter(User.name.in_(['ed', 'wendy']))
for _row in query.all():
print(_row.name, _row.fullname, _row.birth)
print("\n----> NOT IN:")
query = session.query(User).filter(~User.name.in_(['ed', 'wendy']))
for _row in query.all():
print(_row.name, _row.fullname, _row.birth)
print("\n----> AND:")
query = session.query(User).filter(
User.name=='ed', User.fullname=='Ed Jones')
_row = query.first()
print(_row.name, _row.fullname, _row.birth)
print("\n----> OR:")
query = session.query(User).filter(
or_(User.name=='ed', User.name=='wendy'))
for _row in query.all():
print(_row.name, _row.fullname, _row.birth)
print("\n----> NULL:")
query = session.query(User).filter(User.birth == None)
for _row in query.all():
print(_row.name, _row.fullname)
print("\n----> NOT NULL:")
query = session.query(User).filter(User.birth != None)
for _row in query.all():
print(_row.name, _row.fullname)
print("\n----> LIKE")
query = session.query(User).filter(User.name.like('%ed%'))
for _row in query.all():
print(_row.name, _row.fullname)
output:
.. code-block:: bash
----> order_by(id):
ed Ed Jones 1989-07-01 00:00:00
wendy Wendy Williams 1983-04-01 00:00:00
mary Mary Contrary 1990-01-30 00:00:00
fred Fred Flinstone 1977-03-12 00:00:00
justin Justin Bieber None
----> order_by(desc(id)):
justin Justin Bieber None
fred Fred Flinstone 1977-03-12 00:00:00
mary Mary Contrary 1990-01-30 00:00:00
wendy Wendy Williams 1983-04-01 00:00:00
ed Ed Jones 1989-07-01 00:00:00
----> order_by(date):
fred Fred Flinstone 1977-03-12 00:00:00
wendy Wendy Williams 1983-04-01 00:00:00
ed Ed Jones 1989-07-01 00:00:00
mary Mary Contrary 1990-01-30 00:00:00
justin Justin Bieber None
----> EQUAL:
wendy Wendy Williams 1983-04-01 00:00:00
----> NOT EQUAL:
ed Ed Jones 1989-07-01 00:00:00
mary Mary Contrary 1990-01-30 00:00:00
fred Fred Flinstone 1977-03-12 00:00:00
justin Justin Bieber None
----> IN:
ed Ed Jones 1989-07-01 00:00:00
wendy Wendy Williams 1983-04-01 00:00:00
----> NOT IN:
mary Mary Contrary 1990-01-30 00:00:00
fred Fred Flinstone 1977-03-12 00:00:00
justin Justin Bieber None
----> AND:
ed Ed Jones 1989-07-01 00:00:00
----> OR:
ed Ed Jones 1989-07-01 00:00:00
wendy Wendy Williams 1983-04-01 00:00:00
----> NULL:
justin Justin Bieber
----> NOT NULL:
ed Ed Jones
wendy Wendy Williams
mary Mary Contrary
fred Fred Flinstone
----> LIKE
ed Ed Jones
fred Fred Flinstone
mapper: Map ``Table`` to ``class``
-----------------------------------
.. code-block:: python
from sqlalchemy import (
create_engine,
Table,
MetaData,
Column,
Integer,
String,
ForeignKey)
from sqlalchemy.orm import (
mapper,
relationship,
sessionmaker)
# classical mapping: map "table" to "class"
db_url = 'sqlite://'
engine = create_engine(db_url)
meta = MetaData(bind=engine)
user = Table('User', meta,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('fullname', String),
Column('password', String))
addr = Table('Address', meta,
Column('id', Integer, primary_key=True),
Column('email', String),
Column('user_id', Integer, ForeignKey('User.id')))
# map table to class
class User(object):
def __init__(self, name, fullname, password):
self.name = name
self.fullname = fullname
self.password = password
class Address(object):
def __init__(self, email):
self.email = email
mapper(User, user, properties={
'addresses': relationship(Address, backref='user')})
mapper(Address, addr)
# create table
meta.create_all()
# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
u = User(name='Hello', fullname='HelloWorld', password='ker')
a = Address(email='hello@hello.com')
u.addresses.append(a)
try:
session.add(u)
session.commit()
# query result
u = session.query(User).filter(User.name == 'Hello').first()
print(u.name, u.fullname, u.password)
finally:
session.close()
output:
.. code-block:: bash
$ python map_table_class.py
Hello HelloWorld ker
Get table dynamically
----------------------
.. code-block:: python
from sqlalchemy import (
create_engine,
MetaData,
Table,
inspect,
Column,
String,
Integer)
from sqlalchemy.orm import (
mapper,
scoped_session,
sessionmaker)
db_url = "sqlite://"
engine = create_engine(db_url)
metadata = MetaData(engine)
class TableTemp(object):
def __init__(self, name):
self.name = name
def get_table(name):
if name in metadata.tables:
table = metadata.tables[name]
else:
table = Table(name, metadata,
Column('id', Integer, primary_key=True),
Column('name', String))
table.create(engine)
cls = type(name.title(), (TableTemp,), {})
mapper(cls, table)
return cls
# get table first times
t = get_table('Hello')
# get table secone times
t = get_table('Hello')
Session = scoped_session(sessionmaker(bind=engine))
try:
Session.add(t(name='foo'))
Session.add(t(name='bar'))
for _ in Session.query(t).all():
print(_.name)
except Exception as e:
Session.rollback()
finally:
Session.close()
output:
.. code-block:: bash
$ python get_table.py
foo
bar
Object Relational join two tables
----------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
name = Column(String)
addresses = relationship("Address", backref="user")
class Address(Base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True)
email = Column(String)
user_id = Column(Integer, ForeignKey('user.id'))
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
# create engine
engine = create_engine(URL(**db_url))
# create tables
Base.metadata.create_all(bind=engine)
# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
user = User(name='user1')
mail1 = Address(email='user1@foo.com')
mail2 = Address(email='user1@bar.com')
user.addresses.extend([mail1, mail2])
session.add(user)
session.add_all([mail1, mail2])
session.commit()
query = session.query(Address, User).join(User)
for _a, _u in query.all():
print(_u.name, _a.email)
output:
.. code-block:: bash
$ python sqlalchemy_join.py
user1 user1@foo.com
user1 user1@bar.com
join on relationship and group_by count
----------------------------------------
.. code-block:: python
from sqlalchemy import (
create_engine,
Column,
String,
Integer,
ForeignKey,
func)
from sqlalchemy.orm import (
relationship,
sessionmaker,
scoped_session)
from sqlalchemy.ext.declarative import declarative_base
db_url = 'sqlite://'
engine = create_engine(db_url)
Base = declarative_base()
class Parent(Base):
__tablename__ = 'parent'
id = Column(Integer, primary_key=True)
name = Column(String)
children = relationship('Child', back_populates='parent')
class Child(Base):
__tablename__ = 'child'
id = Column(Integer, primary_key=True)
name = Column(String)
parent_id = Column(Integer, ForeignKey('parent.id'))
parent = relationship('Parent', back_populates='children')
Base.metadata.create_all(bind=engine)
Session = scoped_session(sessionmaker(bind=engine))
p1 = Parent(name="Alice")
p2 = Parent(name="Bob")
c1 = Child(name="foo")
c2 = Child(name="bar")
c3 = Child(name="ker")
c4 = Child(name="cat")
p1.children.extend([c1, c2, c3])
p2.children.append(c4)
try:
Session.add(p1)
Session.add(p2)
Session.commit()
# count number of children
q = Session.query(Parent, func.count(Child.id))\
.join(Child)\
.group_by(Parent.id)
# print result
for _p, _c in q.all():
print('parent: {}, num_child: {}'.format(_p.name, _c))
finally:
Session.remove()
output:
.. code-block:: bash
$ python join_group_by.py
parent: Alice, num_child: 3
parent: Bob, num_child: 1
Create tables with dynamic columns (ORM)
------------------------------------------
.. code-block:: python
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String
from sqlalchemy import inspect
from sqlalchemy.engine.url import URL
from sqlalchemy.ext.declarative import declarative_base
db_url = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '192.168.99.100',
'port': 5432}
engine = create_engine(URL(**db_url))
Base = declarative_base()
def create_table(name, cols):
Base.metadata.reflect(engine)
if name in Base.metadata.tables: return
table = type(name, (Base,), cols)
table.__table__.create(bind=engine)
create_table('Table1', {
'__tablename__': 'Table1',
'id': Column(Integer, primary_key=True),
'name': Column(String)})
create_table('Table2', {
'__tablename__': 'Table2',
'id': Column(Integer, primary_key=True),
'key': Column(String),
'val': Column(String)})
inspector = inspect(engine)
for _t in inspector.get_table_names():
print(_t)
output:
.. code-block:: bash
$ python sqlalchemy_dynamic_orm.py
Table1
Table2
Close database connection
--------------------------
.. code-block:: python
from sqlalchemy import (
create_engine,
event,
Column,
Integer)
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite://')
base = declarative_base()
@event.listens_for(engine, 'engine_disposed')
def receive_engine_disposed(engine):
print("engine dispose")
class Table(base):
__tablename__ = 'example table'
id = Column(Integer, primary_key=True)
base.metadata.create_all(bind=engine)
session = sessionmaker(bind=engine)()
try:
try:
row = Table()
session.add(row)
except Exception as e:
session.rollback()
raise
finally:
session.close()
finally:
engine.dispose()
output:
.. code-block:: bash
$ python db_dispose.py
engine dispose
.. warning::
Be careful. Close *session* does not mean close database connection.
SQLAlchemy *session* generally represents the *transactions*, not connections.
Cannot use the object after close the session
-----------------------------------------------
.. code-block:: python
from __future__ import print_function
from sqlalchemy import (
create_engine,
Column,
String,
Integer)
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
url = 'sqlite://'
engine = create_engine(url)
base = declarative_base()
class Table(base):
__tablename__ = 'table'
id = Column(Integer, primary_key=True)
key = Column(String)
val = Column(String)
base.metadata.create_all(bind=engine)
session = sessionmaker(bind=engine)()
try:
t = Table(key="key", val="val")
try:
print(t.key, t.val)
session.add(t)
session.commit()
except Exception as e:
print(e)
session.rollback()
finally:
session.close()
print(t.key, t.val) # exception raise from here
except Exception as e:
print("Cannot use the object after close the session")
finally:
engine.dispose()
output:
.. code-block:: bash
$ python sql.py
key val
Cannot use the object after close the session
Hooks
-----
.. code-block:: python
from sqlalchemy import Column, String, Integer
from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
url = "sqlite:///:memory:"
engine = create_engine(url)
Base.metadata.create_all(bind=engine)
Session = sessionmaker(bind=engine)
@event.listens_for(User, "before_insert")
def before_insert(mapper, connection, user):
print(f"before insert: {user.name}")
@event.listens_for(User, "after_insert")
def after_insert(mapper, connection, user):
print(f"after insert: {user.name}")
try:
session = scoped_session(Session)
user = User(name="bob", age=18)
session.add(user)
session.commit()
except SQLAlchemyError as e:
session.rollback()
finally:
session.close()