Scribd
Upload
31 views38 pages

SQLAlchemy Basics for Python Users

The document provides a comprehensive guide on using SQLAlchemy for database interactions in Python, covering topics such as setting up database URLs, creating and managing tables, executing SQL statements, and performing bulk inserts. It includes code examples for various operations like creating tables, inserting data, selecting records, joining tables, and reflecting existing database schemas. Additionally, it discusses advanced features like using the COPY statement for bulk inserts in PostgreSQL and returning inserted IDs.

Uploaded by

Wadia Alfred
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
31 views38 pages

SQLAlchemy Basics for Python Users

The document provides a comprehensive guide on using SQLAlchemy for database interactions in Python, covering topics such as setting up database URLs, creating and managing tables, executing SQL statements, and performing bulk inserts. It includes code examples for various operations like creating tables, inserting data, selecting records, joining tables, and reflecting existing database schemas. Additionally, it discusses advanced features like using the COPY statement for bulk inserts in PostgreSQL and returning inserted IDs.

Uploaded by

Wadia Alfred
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

SQLAlchemy

[Link]/notes/[Link]

Set a database URL

from [Link] import URL

postgres_db = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'port': 5432}
print(URL(**postgres_db))

sqlite_db = {'drivername': 'sqlite', 'database': '[Link]'}


print(URL(**sqlite_db))

output:

$ python sqlalchemy_url.py
postgres://postgres:postgres@[Link]:5432
sqlite:///[Link]

Sqlalchemy Support DBAPI - PEP249

1/38
from sqlalchemy import create_engine

db_uri = "sqlite:///[Link]"
engine = create_engine(db_uri)

# DBAPI - PEP249
# create table
[Link]('CREATE TABLE "EX1" ('
'id INTEGER NOT NULL,'
'name VARCHAR, '
'PRIMARY KEY (id));')
# insert a raw
[Link]('INSERT INTO "EX1" '
'(id, name) '
'VALUES (1,"raw1")')

# select *
result = [Link]('SELECT * FROM '
'"EX1"')
for _r in result:
print(_r)

# delete *
[Link]('DELETE from "EX1" where id=1;')
result = [Link]('SELECT * FROM "EX1"')
print([Link]())

Transaction and Connect Object

from sqlalchemy import create_engine

db_uri = 'sqlite:///[Link]'
engine = create_engine(db_uri)

# Create connection
conn = [Link]()
# Begin transaction
trans = [Link]()
[Link]('INSERT INTO "EX1" (name) '
'VALUES ("Hello")')
[Link]()
# Close connection
[Link]()

Metadata - Generating Database Schema

2/38
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:///[Link]'
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 [Link]:
print("Table: ", _t)

Inspect - Get Database Information

from sqlalchemy import create_engine


from sqlalchemy import inspect

db_uri = 'sqlite:///[Link]'
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

from sqlalchemy import create_engine


from sqlalchemy import MetaData
from sqlalchemy import Table

db_uri = 'sqlite:///[Link]'
engine = create_engine(db_uri)

# Create a MetaData instance


metadata = MetaData()
print([Link])

# reflect db schema to MetaData


[Link](bind=engine)
print([Link])

3/38
Print Create Table Statement with Indexes (SQL DDL)

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([Link](dialect=[Link]))

meta = MetaData()
example_table = Table('Example',meta,
Column('id', Integer, primary_key=True),
Column('name', String(10), index=True))

db_uri = 'sqlite:///[Link]'
engine = create_engine(db_uri, strategy='mock', executor=metadata_dump)

meta.create_all(bind=engine, tables=[example_table])

output:

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

from sqlalchemy import create_engine


from sqlalchemy import MetaData
from sqlalchemy import Table

db_uri = 'sqlite:///[Link]'
engine = create_engine(db_uri)

# Create MetaData instance


metadata = MetaData(engine).reflect()
print([Link])

# Get Table
ex_table = [Link]['Example']
print(ex_table)

Create all Tables Store in “MetaData”

4/38
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:///[Link]'
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

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:///[Link]'
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))
[Link]()

Create table with same columns

5/38
from sqlalchemy import (
create_engine,
inspect,
Column,
String,
Integer)

from [Link] 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"

[Link].create_all(bind=engine)

# check table exists


ins = inspect(engine)
for _t in ins.get_table_names():
print(_t)

Drop a Table

6/38
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import inspect
from sqlalchemy import Table
from sqlalchemy import Column, Integer, String
from [Link] import URL

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'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))

[Link](engine)
inspector = inspect(engine)
print('Test' in inspector.get_table_names())

[Link](engine)
inspector = inspect(engine)
print('Test' in inspector.get_table_names())

output:

$ python sqlalchemy_drop.py
$ True
$ False

Some Table Object Operation

7/38
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([Link])

# Get Columns
print([Link]())

# Get Column
c = [Link]
print([Link])
# Or
c = [Link]
print([Link])

# Get Table from Column


print([Link])

SQL Expression Language

8/38
# 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([Link] > 3))

# or expression
print(([Link] > 5) | ([Link] < 2))
# Equal to
print(or_([Link] > 5, [Link] < 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([Link] + 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

9/38
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:///[Link]'
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 = [Link]().values(
l_name='Hello',
f_name='World')
conn = [Link]()
[Link](ins)

# insert multiple data


[Link]([Link](),[
{'l_name':'Hi','f_name':'bob'},
{'l_name':'yo','f_name':'alice'}])

select() - Create a “SELECT” Statement

10/38
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import select
from sqlalchemy import or_

db_uri = 'sqlite:///[Link]'
engine = create_engine(db_uri)
conn = [Link]()

meta = MetaData(engine).reflect()
table = [Link]['user']

# select * from 'user'


select_st = select([table]).where(
table.c.l_name == 'Hello')
res = [Link](select_st)
for _row in res:
print(_row)

# or equal to
select_st = [Link]().where(
table.c.l_name == 'Hello')
res = [Link](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 = [Link](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 = [Link](select_st)
for _row in res:
print(_row)

join() - Joined Two Tables via “JOIN” Statement

11/38
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:///[Link]'
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 = [Link]['user']

# insert
conn = [Link]()
[Link](email_t.insert(),[
{'email':'ker@test','name':'Hi'},
{'email':'yo@test','name':'Hello'}])
# join statement
join_obj = user_t.join(email_t,
email_t.[Link] == user_t.c.l_name)
# using select_from
sel_st = select(
[user_t.c.l_name, email_t.[Link]]).select_from(join_obj)
res = [Link](sel_st)
for _row in res:
print(_row)

Fastest Bulk Insert in PostgreSQL via “COPY” Statement

# This method found here:


[Link]
import io
from datetime import date

from [Link] 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',


12/38
'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'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 = [Link]()

# generate rows
for i in range(100):
line = '\t'.join(
[
f'Name {i}', # first_name
str(18 + i), # age
str([Link]()), # birth_day
]
)
[Link](line + '\n')

# reset file to start


[Link](0)

# bulk insert via `COPY` statement


conn = engine.raw_connection()
with [Link]() as cur:
# [Link]
cur.copy_from(
datafile,
[Link], # table name
sep='\t',
columns=('first_name', 'age', 'birth_day'),
)
[Link]()

Bulk PostgreSQL Insert and Return Inserted IDs

13/38
from [Link] 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': '[Link]',
'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 = [Link]().values(data).returning([Link])
# 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 [Link]
for rowid in [Link](stmt).fetchall():
print(rowid['id'])

output:

$ python sqlalchemy_bulk.py
1
2
3
4
5
6
7
8
9
10

Update Multiple Rows


14/38
from [Link] 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 [Link] import bindparam

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'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 = [Link]().where([Link] == 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
[Link] = %(_id)s

[Link](stmt, data)

Delete Rows from Table

15/38
from sqlalchemy import create_engine
from sqlalchemy import MetaData

db_uri = 'sqlite:///[Link]'
engine = create_engine(db_uri)
conn = [Link]()

meta = MetaData(engine).reflect()
user_t = [Link]['user']

# select * from user_t


sel_st = user_t.select()
res = [Link](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 = [Link](del_st)

# check rows has been delete


sel_st = user_t.select()
res = [Link](sel_st)
for _row in res:
print(_row)

Check Table Existing

16/38
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Column
from sqlalchemy import Integer, String
from sqlalchemy import inspect
from [Link] 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:///[Link]'
engine = create_engine(db_uri)
[Link].create_all(engine)

# check register table exist to Modal


for _t in [Link]:
print(_t)

# check all table in database


meta = MetaData(engine).reflect()
for _t in [Link]:
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

17/38
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import inspect
from sqlalchemy import Column, String, Integer
from [Link] import URL

db = {'drivername': 'postgres',
'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'port': 5432}

url = URL(**db)
engine = create_engine(url)

metadata = MetaData()
[Link](bind=engine)

def create_table(name, metadata):


tables = [Link]()
if name not in tables:
table = Table(name, metadata,
Column('id', Integer, primary_key=True),
Column('key', String),
Column('val', Integer))
[Link](engine)

tables = ['table1', 'table2', 'table3']


for _t in tables: create_table(_t, metadata)

inspector = inspect(engine)
print(inspector.get_table_names())

output:

$ python sqlalchemy_create.py
[u'table1', u'table2', u'table3']

Create tables with dynamic columns (Table)

18/38
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String
from sqlalchemy import Table
from sqlalchemy import MetaData
from sqlalchemy import inspect
from [Link] import URL

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'port': 5432}

engine = create_engine(URL(**db_url))

def create_table(name, *cols):


meta = MetaData()
[Link](bind=engine)
if name in [Link]: return

table = Table(name, meta, *cols)


[Link](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:

$ python sqlalchemy_dynamic.py
Table1
Table2

Object Relational add data

19/38
from datetime import datetime

from sqlalchemy import create_engine


from sqlalchemy import Column, Integer, String, DateTime
from [Link] import sessionmaker
from [Link] import SQLAlchemyError
from [Link] import declarative_base
from [Link] import URL

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'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=[Link])

# create tables
[Link].create_all(bind=engine)

# create session
Session = sessionmaker()
[Link](bind=engine)
session = Session()

data = {'a': 5566, 'b': 9527, 'c': 183}


try:
for _key, _val in [Link]():
row = TestTable(key=_key, val=_val)
[Link](row)
[Link]()
except SQLAlchemyError as e:
print(e)
finally:
[Link]()

Object Relational update data

20/38
from datetime import datetime

from sqlalchemy import create_engine


from sqlalchemy import Column, Integer, String, DateTime
from [Link] import sessionmaker
from [Link] import SQLAlchemyError
from [Link] import declarative_base
from [Link] import URL

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'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=[Link])

# create tables
[Link].create_all(bind=engine)

# create session
Session = sessionmaker()
[Link](bind=engine)
session = Session()

try:
# add row to database
row = TestTable(key="hello", val="world")
[Link](row)
[Link]()

# update row to database


row = [Link](TestTable).filter(
[Link] == 'hello').first()
print('original:', [Link], [Link])
[Link] = "Hello"
[Link] = "World"
[Link]()

# check update correct


row = [Link](TestTable).filter(
[Link] == 'Hello').first()
print('update:', [Link], [Link])
except SQLAlchemyError as e:
print(e)
finally:
[Link]()

21/38
output:

$ python sqlalchemy_update.py
original: hello world
update: Hello World

Object Relational delete row

from datetime import datetime

from sqlalchemy import create_engine


from sqlalchemy import Column, Integer, String, DateTime
from [Link] import sessionmaker
from [Link] import SQLAlchemyError
from [Link] import declarative_base
from [Link] import URL

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'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=[Link])

# create tables
[Link].create_all(bind=engine)

# create session
Session = sessionmaker()
[Link](bind=engine)
session = Session()

row = TestTable(key='hello', val='world')


[Link](row)
query = [Link](TestTable).filter(
[Link]=='hello')
print([Link]())
[Link]()
query = [Link](TestTable).filter(
[Link]=='hello')
print([Link]())

output:

22/38
$ python sqlalchemy_delete.py
<__main__.TestTable object at 0x104eb8f50>
[]

Object Relational relationship

from sqlalchemy import Column, String, Integer, ForeignKey


from [Link] import relationship
from [Link] 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('[Link]'))

u1 = User()
a1 = Address()
print([Link])
print([Link])

[Link](a1)
print([Link])
print([Link])

output:

$ python sqlalchemy_relationship.py
[]
None
[<__main__.Address object at 0x10c4edb50>]
<__main__.User object at 0x10c4ed810>

Object Relational self association

23/38
import json

from sqlalchemy import (


Column,
Integer,
String,
ForeignKey,
Table)

from [Link] import (


sessionmaker,
relationship)

from [Link] import declarative_base

base = declarative_base()

association = Table("Association", [Link],


Column('left', Integer, ForeignKey('[Link]'), primary_key=True),
Column('right', Integer, ForeignKey('[Link]'), primary_key=True))

class Node(base):
__tablename__ = 'node'
id = Column(Integer, primary_key=True)
label = Column(String)
friends = relationship('Node',
secondary=association,
primaryjoin=id==[Link],
secondaryjoin=id==[Link],
backref='left')
def to_json(self):
return dict(id=[Link],
friends=[_.label for _ in [Link]])

nodes = [Node(label='node_{}'.format(_)) for _ in range(0, 3)]


nodes[0].[Link]([nodes[1], nodes[2]])
nodes[1].[Link](nodes[2])

print('----> right')
print([Link]([_.to_json() for _ in nodes], indent=2))

print('----> left')
print([Link]([_n.to_json() for _n in nodes[1].left], indent=2))

output:

24/38
----> 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

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 [Link] import sessionmaker
from [Link] import SQLAlchemyError
from [Link] import declarative_base
from [Link] import URL

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'port': 5432}

Base = declarative_base()

class User(Base):
__tablename__ = 'User'
25/38
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))
[Link].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()
[Link](bind=engine)
session = Session()

# add_all
session.add_all(users)
[Link]()

print("----> order_by(id):")
query = [Link](User).order_by([Link])
for _row in [Link]():
print(_row.name, _row.fullname, _row.birth)

print("\n----> order_by(desc(id)):")
query = [Link](User).order_by(desc([Link]))
for _row in [Link]():
print(_row.name, _row.fullname, _row.birth)

print("\n----> order_by(date):")
query = [Link](User).order_by([Link])
for _row in [Link]():
print(_row.name, _row.fullname, _row.birth)

print("\n----> EQUAL:")
query = [Link](User).filter([Link] == 2)
_row = [Link]()
print(_row.name, _row.fullname, _row.birth)

26/38
print("\n----> NOT EQUAL:")
query = [Link](User).filter([Link] != 2)
for _row in [Link]():
print(_row.name, _row.fullname, _row.birth)

print("\n----> IN:")
query = [Link](User).filter([Link].in_(['ed', 'wendy']))
for _row in [Link]():
print(_row.name, _row.fullname, _row.birth)

print("\n----> NOT IN:")


query = [Link](User).filter(~[Link].in_(['ed', 'wendy']))
for _row in [Link]():
print(_row.name, _row.fullname, _row.birth)

print("\n----> AND:")
query = [Link](User).filter(
[Link]=='ed', [Link]=='Ed Jones')
_row = [Link]()
print(_row.name, _row.fullname, _row.birth)

print("\n----> OR:")
query = [Link](User).filter(
or_([Link]=='ed', [Link]=='wendy'))
for _row in [Link]():
print(_row.name, _row.fullname, _row.birth)

print("\n----> NULL:")
query = [Link](User).filter([Link] == None)
for _row in [Link]():
print(_row.name, _row.fullname)

print("\n----> NOT NULL:")


query = [Link](User).filter([Link] != None)
for _row in [Link]():
print(_row.name, _row.fullname)

print("\n----> LIKE")
query = [Link](User).filter([Link]('%ed%'))
for _row in [Link]():
print(_row.name, _row.fullname)

output:

----> 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

27/38
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

from sqlalchemy import (


create_engine,
Table,
28/38
MetaData,
Column,
Integer,
String,
ForeignKey)

from [Link] 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('[Link]')))

# map table to class


class User(object):
def __init__(self, name, fullname, password):
[Link] = name
[Link] = fullname
[Link] = password

class Address(object):
def __init__(self, email):
[Link] = email

mapper(User, user, properties={


'addresses': relationship(Address, backref='user')})
mapper(Address, addr)

# create table
meta.create_all()

# create session
Session = sessionmaker()
[Link](bind=engine)
session = Session()

u = User(name='Hello', fullname='HelloWorld', password='ker')


a = Address(email='hello@[Link]')
[Link](a)
29/38
try:
[Link](u)
[Link]()

# query result
u = [Link](User).filter([Link] == 'Hello').first()
print([Link], [Link], [Link])

finally:
[Link]()

output:

$ python map_table_class.py
Hello HelloWorld ker

Get table dynamically

30/38
from sqlalchemy import (
create_engine,
MetaData,
Table,
inspect,
Column,
String,
Integer)

from [Link] import (


mapper,
scoped_session,
sessionmaker)

db_url = "sqlite://"
engine = create_engine(db_url)
metadata = MetaData(engine)

class TableTemp(object):
def __init__(self, name):
[Link] = name

def get_table(name):
if name in [Link]:
table = [Link][name]
else:
table = Table(name, metadata,
Column('id', Integer, primary_key=True),
Column('name', String))
[Link](engine)

cls = type([Link](), (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:
[Link](t(name='foo'))
[Link](t(name='bar'))
for _ in [Link](t).all():
print(_.name)
except Exception as e:
[Link]()
finally:
[Link]()

output:

31/38
$ python get_table.py
foo
bar

Object Relational join two tables

32/38
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, ForeignKey
from [Link] import relationship
from [Link] import URL
from [Link] import sessionmaker
from [Link] 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('[Link]'))

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'port': 5432}

# create engine
engine = create_engine(URL(**db_url))

# create tables
[Link].create_all(bind=engine)

# create session
Session = sessionmaker()
[Link](bind=engine)
session = Session()

user = User(name='user1')
mail1 = Address(email='user1@[Link]')
mail2 = Address(email='user1@[Link]')
[Link]([mail1, mail2])

[Link](user)
session.add_all([mail1, mail2])
[Link]()

query = [Link](Address, User).join(User)


for _a, _u in [Link]():
print(_u.name, _a.email)

output:

33/38
$ python sqlalchemy_join.py
user1 user1@[Link]
user1 user1@[Link]

join on relationship and group_by count

from sqlalchemy import (


create_engine,
Column,
String,
Integer,
ForeignKey,
func)

from [Link] import (


relationship,
sessionmaker,
scoped_session)

from [Link] 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('[Link]'))
parent = relationship('Parent', back_populates='children')

[Link].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")

[Link]([c1, c2, c3])


[Link](c4)

try:
34/38
[Link](p1)
[Link](p2)
[Link]()

# count number of children


q = [Link](Parent, [Link]([Link]))\
.join(Child)\
.group_by([Link])

# print result
for _p, _c in [Link]():
print('parent: {}, num_child: {}'.format(_p.name, _c))
finally:
[Link]()

output:

$ python join_group_by.py
parent: Alice, num_child: 3
parent: Bob, num_child: 1

Create tables with dynamic columns (ORM)

35/38
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String
from sqlalchemy import inspect
from [Link] import URL
from [Link] import declarative_base

db_url = {'drivername': 'postgres',


'username': 'postgres',
'password': 'postgres',
'host': '[Link]',
'port': 5432}

engine = create_engine(URL(**db_url))
Base = declarative_base()

def create_table(name, cols):


[Link](engine)
if name in [Link]: 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:

$ python sqlalchemy_dynamic_orm.py
Table1
Table2

Close database connection

36/38
from sqlalchemy import (
create_engine,
event,
Column,
Integer)

from [Link] import sessionmaker


from [Link] 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)

[Link].create_all(bind=engine)
session = sessionmaker(bind=engine)()

try:
try:
row = Table()
[Link](row)
except Exception as e:
[Link]()
raise
finally:
[Link]()
finally:
[Link]()

output:

$ 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

37/38
from __future__ import print_function

from sqlalchemy import (


create_engine,
Column,
String,
Integer)

from [Link] import sessionmaker


from [Link] 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)

[Link].create_all(bind=engine)
session = sessionmaker(bind=engine)()

try:
t = Table(key="key", val="val")
try:
print([Link], [Link])
[Link](t)
[Link]()
except Exception as e:
print(e)
[Link]()
finally:
[Link]()

print([Link], [Link]) # exception raise from here


except Exception as e:
print("Cannot use the object after close the session")
finally:
[Link]()

output:

$ python [Link]
key val
Cannot use the object after close the session

38/38

Common questions

Powered by AI

Data insertion in SQLAlchemy uses the insert() construct on a table object, allowing for the addition of values via the values() method. Multiple entries can be inserted by passing a list of dictionaries to the engine execute call or using a similar ORM approach to add multiple objects in one transaction, optimizing database interactions by reducing round-trips .

A bulk update in SQLAlchemy involves using the update() method on a table object, passing a dictionary of new values and binding parameters for conditions (`where` clauses). This operation is reflected in SQL with an UPDATE statement specifying the SET clause with the new values and a WHERE clause to locate rows. The SQL generated includes bind parameters to ensure secure updating .

To create a new table using SQLAlchemy, you define the table using the Table class, specifying columns and primary keys. You then use MetaData to bind to an engine and call create_all() to create the table. To inspect the table's existence, use the inspect(engine) to get its table names and check if the table exists in this list. This approach utilizes both the SQLAlchemy ORM for table modeling and reflection functionalities for database inspection .

SQLAlchemy supports bulk data insertion into PostgreSQL through the `COPY` statement, accessed via raw connection to leverage PostgreSQL's high-speed data loading utilities. This method significantly improves performance by reducing the overhead of individual inserts, making it suitable for large datasets and reducing server round-trips while ensuring transaction safety .

SQLAlchemy facilitates table joins using the join() method on table objects and defining conditions for joining, often via foreign key relationships. This enables complex queries across multiple tables in a single SQL statement, enhancing efficiency and reducing the need for handling separate datasets in Python. It is particularly beneficial in ORM contexts where object relationships can be traversed naturally .

SQLAlchemy enhances schema management flexibility by allowing tables to be created dynamically with varying column definitions, using reflection to check existing tables and only create those not present. This feature supports altering database schemas on-the-fly, adapting to evolving application needs without manual database updates, thus facilitating agile development practices .

SQLAlchemy uses declarative base classes to establish foreign key relationships via ForeignKey and relationship attributes. While they allow complex entity navigation and ORM mapping, developers must ensure consistency in foreign key constraints and handle potential circular dependencies carefully to avoid performance issues or constraint violations during operations .

Session management in SQLAlchemy ORM plays a critical role in handling transactions as it serves as a workspace for all queries and persistent operations. It ensures changes are committed as a whole, providing rollback capabilities in case of errors, thereby maintaining database integrity. This transactional nature aids performance by allowing bulk transactions rather than individual row-level operations, reducing the overhead of database communication .

SQLAlchemy ORM handles relationships between tables using backref and relationship attributes. This enables navigation between related objects in application code, allowing for lazy loading of related objects and cascading operations like updates and deletes. It provides a high level of abstraction, allowing developers to work with Python objects while automatically managing underlying SQL operations .

SQLAlchemy's reflection features interact with databases by querying the database schema to load metadata dynamically into Python. This approach offers significant advantages, such as automating schema synchronization between the database and ORM models, facilitating migrations, and ensuring that application code does not become stale relative to database structures. This automation reduces maintenance and fixes errors due to schema drift .

You might also like