SQLAlchemy Cheatsheet (For PostgreSQL)

发表于 阅读次数： Disqus：

前言

最近在用SQLAlchemy处理数据库相关的事情，在网上找到了一篇关于SQLAlchemy基本用法的Cheatsheet，自己跑了一下，发现其中还有一些Bugs，做了一点改正，记录下来。

注意

文中的所有用法以PostgreSQL为例，并进行测试。对于其他类型的数据库，个别特殊的地方可能需要修改。

Set a database URL

1
from sqlalchemy.engine.url import URL
1
2
3
4
5
6
7
8
9
postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}
db_url = URL(**postgres_db)
print(db_url)
postgres+psycopg2://postgres:postgres@localhost:5432/example

1
from sqlalchemy import create_engine

Note that the Engine and its underlying Pool do not establish the first actual DBAPI connection until the Engine.connect() method is called, or an operation which is dependent on this method such as Engine.execute() is invoked. In this way, Engine and Pool can be said to have a lazy initialization behavior.

Create a Database (For postgresql only)

方法一：利用底层连接

在Postgres上，默认情况下有三个数据库。如果您能够以超级用户的身份进行连接(例如，postgres)，则可以连接到postgres或template1数据库。默认pg_conf只允许名为postgres使用postgres角色，所以最简单的事情就是成为那个用户。无论如何，与通常一样，与具有创建数据库权限的用户一起创建一个引擎：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
no_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': None,
               'query': None}
no_db_url = URL(**no_db)
engine = create_engine(no_db_url)
# 因为Postgres不允许在事务中创建数据库，
# sqlalchemy总是尝试在事务中运行查询。
# 为了避免这种情况，从引擎获取底层连接

# 连接仍然在事务中，所以必须用commit
conn = engine.connect()
conn.execute("commit")
conn.execute("create database example")
conn.close()
engine.dispose()

方法二：采用sqlalchemy_utils

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# refer to: http://sqlalchemy-utils.readthedocs.org/en/latest/database_helpers.html

from sqlalchemy_utils import database_exists, create_database,drop_database
no_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'no_db',
               'query': None}
no_db_url = URL(**no_db)
engine = create_engine(no_db_url)
print(database_exists(engine.url))

if database_exists(engine.url):
    drop_database(no_db_url)
    print(database_exists(engine.url))

if not database_exists(engine.url):
    create_database(engine.url)
print(database_exists(engine.url))

if database_exists(engine.url):
    drop_database(no_db_url)
print(database_exists(engine.url))

engine.dispose()
False
True
False

Sqlalchemy Support DBAPI - PEP249

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from sqlalchemy import create_engine

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

# 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())
engine.dispose()
(1, 'raw1')
[]

Transaction and Connect Object

单引号 (') 和双引号 (") 在 PostgreSQL 中有明确的区分，单引号用来引用值 (value)，而双引号用来引用标识符，如字段名，表名等等。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from sqlalchemy import create_engine

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

# Create connection
conn = engine.connect()
# Begin transaction
trans = conn.begin()
conn.execute('INSERT INTO "EX1" (id, name) '
             'VALUES (2, \'Hello\')')
trans.commit()
# Close connection
conn.close()
engine.dispose()

Metadata - Generating Database Schema

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer, String

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

# Create a metadata instance
metadata = MetaData(engine)
# Declare a table
table = Table('SubExample',metadata,
              Column('id',Integer, primary_key=True),
              Column('name',String))
# Create all tables
metadata.create_all()
for _t in metadata.tables:
   print("Table: ", _t)
Table:  SubExample

Inspect - Get Database Information

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from sqlalchemy import create_engine
from sqlalchemy import inspect

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)
inspector = inspect(engine)

# Get table information
print(inspector.get_table_names())

# Get column information
print(inspector.get_columns('EX1'))
print([item['name'] for item in inspector.get_columns('SubExample')])
['EX1', 'SubExample']
[{'name': 'id', 'type': INTEGER(), 'nullable': False, 'default': None, 'autoincrement': False, 'comment': None}, {'name': 'name', 'type': VARCHAR(), 'nullable': True, 'default': None, 'autoincrement': False, 'comment': None}]
['id', 'name']

Reflection - Loading Table from Existing Database

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

# Create a MetaData instance
metadata = MetaData()
print(metadata.tables)

# reflect db schema to MetaData
metadata.reflect(bind=engine)
print(metadata.tables)
immutabledict({})
immutabledict({'EX1': Table('EX1', MetaData(bind=None), Column('id', INTEGER(), table=<EX1>, primary_key=True, nullable=False), Column('name', VARCHAR(), table=<EX1>), schema=None), 'SubExample': Table('SubExample', MetaData(bind=None), Column('id', INTEGER(), table=<SubExample>, primary_key=True, nullable=False, server_default=DefaultClause(<sqlalchemy.sql.elements.TextClause object at 0x000001AEF1F68370>, for_update=False)), Column('name', VARCHAR(), table=<SubExample>), schema=None)})

Get Table from MetaData

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

# Create MetaData instance
metadata = MetaData(bind=engine)
metadata.reflect()

# Get Table
table = metadata.tables['EX1']
print(table.columns)
['EX1.id', 'EX1.name']

Create all Tables Store in “MetaData”

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

# Create a MetaData instance
metadata = MetaData(engine)


# Register t1, t2 to metadata
t1 = Table('EX1', metadata,
           Column('id',Integer, primary_key=True),
           Column('name',String))

t2 = Table('EX2', metadata,
           Column('id',Integer, primary_key=True),
           Column('val',Integer))
# Create all tables in metadata
metadata.create_all()
engine.dispose()

Create Specific Table

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

# Create a MetaData instance
metadata = MetaData(engine)
# Register t3 to metadata
t3 = Table('EX3', metadata,
           Column('id',Integer, primary_key=True),
           Column('name',String))
t3.create()
engine.dispose()

Create table with same columns

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from sqlalchemy import (
    create_engine,
    inspect,
    Column,
    String,
    Integer)
from sqlalchemy.ext.declarative import declarative_base

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
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
inspector = inspect(engine)
print(inspector.get_table_names())
engine.dispose()
['EX1', 'SubExample', 'EX2', 'EX3', 'downtown_a_people', 'downtown_b_people']

Drop a table

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from sqlalchemy import (
    create_engine,
    inspect,
    Column,
    String,
    Integer)
from sqlalchemy.ext.declarative import declarative_base

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

metadata = MetaData()
table = Table('Test', metadata,
              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())

engine.dispose()
True
False

Some Table Object Operation

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from sqlalchemy import (
    create_engine,
    inspect,
    Column,
    String,
    Integer)
from sqlalchemy.ext.declarative import declarative_base

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

metadata = MetaData()

t = Table('ex_table', metadata,
          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)

engine.dispose()
ex_table
['id', 'key', 'val']
key
key
ex_table

SQL Expression Language

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
# 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_

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

metadata = MetaData()

table = Table('example', metadata,
              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']))

engine.dispose()
<sqlalchemy.sql.elements.BinaryExpression object at 0x000001AEF238F100>
example.l_name = :l_name_1
<sqlalchemy.sql.elements.BinaryExpression object at 0x000001AEF2297D90>
<sqlalchemy.sql.elements.BinaryExpression object at 0x000001AEF234D4C0>
example.id > :id_1 OR example.id < :id_2
example.id > :id_1 OR example.id < :id_2
example.l_name IS NULL
example.l_name IS NULL
example.id + :id_1
example.l_name || :l_name_1
example.l_name IN (:l_name_1, :l_name_2)

insert() - Create an “INSERT” Statement

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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


postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

metadata = MetaData(engine)

# create table
table = Table('user', metadata,
   Column('id', Integer, primary_key=True),
   Column('l_name', String),
   Column('f_name', String))
metadata.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'}])

conn.close()
engine.dispose()

select() - Create a “SELECT” Statement

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import select
from sqlalchemy import or_


postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)
conn = engine.connect()

metadata = MetaData(engine)
metadata.reflect()

table = metadata.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)
    
conn.close()
engine.dispose()
(1, 'Hello', 'World')
(1, 'Hello', 'World')
('Hello', 'World')
('Hi', 'bob')
(2, 'Hi', 'bob')
(1, 'Hello', 'World')

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)
conn = engine.connect()

metadata = MetaData(engine)
metadata.reflect()
email_t = Table('email_addr', metadata,
      Column('id', Integer, primary_key=True),
      Column('email',String),
      Column('name',String))
metadata.create_all()

# get user table
user_t = metadata.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)

conn.close()
engine.dispose()
('Hello', 'yo@test')
('Hi', 'ker@test')

Delete Rows from Table

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from sqlalchemy import create_engine
from sqlalchemy import MetaData

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)
conn = engine.connect()

metadata = MetaData(engine)
metadata.reflect()

user_t = metadata.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)

conn.close()
engine.dispose()
(1, 'Hello', 'World')
(2, 'Hi', 'bob')
(3, 'yo', 'alice')
----- delete -----
(2, 'Hi', 'bob')
(3, 'yo', 'alice')

Check Table Existing

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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))

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)
conn = engine.connect()

# check register table exist to Modal
for _t in Modal.metadata.tables:
    print(_t)
print('-'*50)

# check all table in database
metadata = MetaData(engine)
metadata.reflect()
for _t in metadata.tables:
    print(_t)
print('-'*50)

Modal.metadata.create_all(engine)
# check table names exists via inspect
ins = inspect(engine)
for _t in ins.get_table_names():
    print(_t)
    
conn.close()
engine.dispose()
ex_t
--------------------------------------------------
EX1
user
SubExample
email_addr
EX2
EX3
downtown_a_people
downtown_b_people
--------------------------------------------------
EX1
user
SubExample
email_addr
EX2
ex_t
EX3
downtown_a_people
downtown_b_people

Create multiple tables at once

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)
conn = engine.connect()

metadata = MetaData(engine)
metadata.reflect()

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

conn.close()
engine.dispose()
['EX1', 'user', 'SubExample', 'email_addr', 'EX2', 'ex_t', 'EX3', 'table1', 'table2', 'table3', 'downtown_a_people', 'downtown_b_people']

Create tables with dynamic columns (Table)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

def create_table(name, *cols):
    metadata = MetaData(bind=engine)
    metadata.reflect()
    if name in metadata.tables: return

    table = Table(name, metadata, *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)

engine.dispose()
EX1
user
SubExample
email_addr
EX2
ex_t
EX3
table1
table2
table3
downtown_a_people
downtown_b_people
Table1
Table2

Object Relational add data

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(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()

engine.dispose()

Object Relational update data

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(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()
    
engine.dispose()
original: hello world
update: Hello World

Object Relational delete row

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(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().id)
query.delete()

query = session.query(TestTable).filter(
        TestTable.key=='hello')
session.commit()
print([q.id for q in query.all()])


session.close()
engine.dispose()
5
[]

Object Relational relationship

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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'))

a1 = Address(email='[email protected]')
u1 = User(name='user2')
print(u1.addresses)
print(a1.user)

u1.addresses.append(a1)
print([a.email for a in u1.addresses])
print(a1.user.name)
[]
None
['[email protected]']
user2

Object Relational self association

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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))
--------------------> right
[
  {
    "id": null,
    "friends": [
      "node_1",
      "node_2"
    ]
  },
  {
    "id": null,
    "friends": [
      "node_2"
    ]
  },
  {
    "id": null,
    "friends": []
  }
]
--------------------> left
[
  {
    "id": null,
    "friends": [
      "node_1",
      "node_2"
    ]
  }
]

Object Relational basic query

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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


postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

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
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()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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)

session.close()
engine.dispose()
----> 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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
from sqlalchemy import (
    create_engine,
    Table,
    MetaData,
    Column,
    Integer,
    String,
    ForeignKey,
    inspect)

from sqlalchemy.orm import (
    mapper,
    relationship,
    sessionmaker)

# classical mapping: map "table" to "class"

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

metadata = MetaData(bind=engine)

inspector = inspect(engine)
metadata.reflect()
if 'Address' in inspector.get_table_names():
    # Drop table in the database and remove it from metadata
    metadata.tables['Address'].drop(engine)
    metadata.remove(metadata.tables['Address'])
if 'User' in inspector.get_table_names():
    metadata.reflect()
    # Drop table in the database and remove it from metadata
    metadata.tables['User'].drop(engine)
    metadata.remove(metadata.tables['User'])
    

user = Table('User', metadata,
             Column('id', Integer, primary_key=True),
             Column('name', String),
             Column('fullname', String),
             Column('password', String))

addr = Table('Address', metadata,
             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
metadata.create_all()

# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()

u = User(name='Hello', fullname='HelloWorld', password='ker')
a = Address(email='[email protected]')
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()
    engine.dispose()
Hello HelloWorld ker

Get table dynamically

使用type()函数创建类对象

  1. 使用type()函数时，如果只传入一个参数object，那么将返回该object的类型；
  2. 如果分别传入name，bases，dict这三个参数，那么type()函数将会创建一个对象；
  3. 使用class定义对象的时候，Python解释器调用type()函数来动态创建对象。

要动态创建一个class对象,type()函数依次传入3个参数：

  1. class的名称，字符串形式；
  2. 继承的父类集合，注意Python支持多重继承，如果只有一个父类，注意tuple的单元素写法；
  3. class的方法名称与函数绑定
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
from sqlalchemy import (
    create_engine,
    MetaData,
    Table,
    inspect,
    Column,
    String,
    Integer)

from sqlalchemy.orm import (
    mapper,
    scoped_session,
    sessionmaker)


postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

metadata = MetaData(bind=engine)

class TableTemp(object):
    def __init__(self, name):
        self.name = name
        
def get_table(name):
    metadata.reflect()
    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 class first times
t = get_table('Hello')

# get table class secone times
t = get_table('Hello')
print(t(name='foo'))
Session = scoped_session(sessionmaker(bind=engine))
try:
    Session.add(t(name='foo'))
    Session.add(t(name='bar'))
    Session.commit()
    for _ in Session.query(t).all():
        print(_.name)
except Exception as e:
    Session.rollback()
finally:
    Session.close()
    engine.dispose()
<__main__.Hello object at 0x000001AEF24AA280>
foo
bar

Object Relational join two tables

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from sqlalchemy import create_engine, inspect
from sqlalchemy import Column, Integer, String, ForeignKey, MetaData
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'))
    
postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

# create engine
db_url = URL(**postgres_db)
engine = create_engine(db_url)
metadata = MetaData(bind=engine)
metadata.reflect()
# # check tables existing
inspector = inspect(engine)
if 'address' in inspector.get_table_names():
    metadata.tables['address'].drop(engine)
if 'user' in inspector.get_table_names():
    metadata.tables['user'].drop(engine)

# 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='[email protected]')
mail2 = Address(email='[email protected]')
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)
    
session.close()
engine.dispose()
user1 [email protected]
user1 [email protected]

join on relationship and group_by count

关于反向引用 backref 与 back_populates

SQLAlchemy 提供了 backref 和 back_populates 两个参数。两个参数的效果完全一致，区别在于:

  • backref 只需要在 Parent 类中声明 children，Child.parent 会被动态创建;
  • back_populates 必须在两个类中显式地使用back_populates参数，略显繁琐，但是更加清晰。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

metadata = MetaData(bind=engine)

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()
    engine.dispose()
parent: Bob, num_child: 1
parent: Alice, num_child: 3

Create tables with dynamic columns (ORM)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

metadata = MetaData(bind=engine)
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)

engine.dispose()
EX1
Hello
parent
child
SubExample
email_addr
Test Table
EX2
ex_t
EX3
table1
table2
table3
user
address
User
Address
downtown_a_people
downtown_b_people
Table1
Table2

Close database connection

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from sqlalchemy import (
    create_engine,
    event,
    Column,
    Integer)

from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_url)

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)
        session.commit()
    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()
finally:
    engine.dispose()
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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


postgres_db = {'drivername': 'postgres+psycopg2',
               'username': 'postgres',
               'password': 'postgres',
               'host': 'localhost',
               'port': 5432,
               'database': 'example',
               'query': None}

db_url = URL(**postgres_db)
engine = create_engine(db_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()
key val
Cannot use the object after close the session

小结

里面的很多用法，我自己也还没有用到过，还处在摸索学习的阶段。接下来有这么几个打算：

  1. 梳理一下SQLAlchemy中的基本概念，尤其是把数据库本身的跟ORM的理清楚；
  2. 对前面提到的这些用法进行归类、整理，弄清其中的一些技巧；
  3. 对其中经常用的一些功能，根据自己的需要，再做一下封装。

参考资料

  1. https://www.pythonsheets.com/notes/python-sqlalchemy.html
  2. https://www.cnblogs.com/kirito-c/p/10900024.html
  3. http://einverne.github.io/post/2016/08/sqlalchemy-usage.html#declare-a-mapping
  4. https://docs.sqlalchemy.org/en/13/core/metadata.html
  5. https://stackoverflow.com/questions/41887344/how-to-verify-sqlalchemy-engine-object
  6. https://stackoverflow.com/questions/35918605/how-to-delete-a-table-in-sqlalchemy