SQLAlchemy备忘录

创建数据库URL

from sqlalchemy.engine.url import URL

postgres_db = {
    'drivername':'postgres',
    'username':'postgres',
    'password':'postgres',
    'host':'localhost',
    'port':5432
}
print(URL(**postgres_db))
postgres://postgres:postgres@localhost:5432
sqlite_db = {'drivername':'sqlite', 'database':'db.sqlite'}
print(URL(**sqlite_db))
sqlite:///db.sqlite

支持DBAPI – PEP249

from sqlalchemy import create_engine

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

# 建表
engine.execute('CREATE TABLE example('
              'id Integer NOT NULL,'
              'name varchar, '
              'PRIMARY KEY(id))')
# 插入数据
engine.execute('INSERT INTO example (id, name) VALUES (?, ?)', (1, '小明'))
# 查询
# 没有使用cursor, 直接用变量result接收查询结果
result = engine.execute('SELECT * FROM example')
for r_ in result:
    print(r_)
(1, '小明')
# 删除
engine.execute('DELETE FROM example WHERE id = ?', (1,))
result = result = engine.execute('SELECT * FROM example')
print(result.fetchall())
[]

事务和连接对象

Transaction and Connect Object

from sqlalchemy import create_engine

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

# 创建连接对象
con = engine.connect()

# 开始事务
trans = con.begin()
con.execute('INSERT INTO example(name) VALUES (?)', '小花')

# 提交事务
trans.commit()

# 关闭连接
con.close()

MetaData生成数据库模式

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)

# 创建MetaData实例
meta = MetaData(engine)

# 定义表
book = Table('Book', meta,
             Column('id', Integer, primary_key=True),
             Column('title', String(50)),
             Column('author', String(20)))

author = Table('Author', meta,
             Column('id', Integer, primary_key=True),
             Column('name', String(20)),
             Column('bio', String(50)))

# 创建所有表
meta.create_all()

# 创建指定表
# book.create()

for _t in meta.tables:
    print('Table:', _t)
Table: Book
Table: Author

inspect 获取数据库信息

from sqlalchemy import create_engine
from sqlalchemy import inspect

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

inspector = inspect(engine)

# 获取表信息
print(inspector.get_table_names())

# 获取指定表的字段信息
print(inspector.get_columns('book'))
['Author', 'Book', 'example']
[{'name': 'id', 'type': INTEGER(), 'nullable': False, 'default': None, 'autoincrement': 'auto', 'primary_key': 1}, {'name': 'title', 'type': VARCHAR(length=50), 'nullable': True, 'default': None, 'autoincrement': 'auto', 'primary_key': 0}, {'name': 'author', 'type': VARCHAR(length=20), 'nullable': True, 'default': None, 'autoincrement': 'auto', 'primary_key': 0}]

Reflection – 载入已有数据库到Table

from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

meta = MetaData(engine)
print(meta.tables)
immutabledict({})
meta.reflect(engine) # meta.reflect(bind=engine)
print(meta.tables)
immutabledict({'Author': Table('Author', MetaData(bind=Engine(sqlite:///db.sqlite)), Column('id', INTEGER(), table=, primary_key=True, nullable=False), Column('name', VARCHAR(length=20), table=), Column('bio', VARCHAR(length=50), table=), schema=None), 'Book': Table('Book', MetaData(bind=Engine(sqlite:///db.sqlite)), Column('id', INTEGER(), table=, primary_key=True, nullable=False), Column('title', VARCHAR(length=50), table=), Column('author', VARCHAR(length=20), table=), schema=None), 'example': Table('example', MetaData(bind=Engine(sqlite:///db.sqlite)), Column('id', INTEGER(), table=, primary_key=True, nullable=False), Column('name', VARCHAR(), table=), schema=None)})

从MetaData获取Table

from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

meta = MetaData(engine)
meta.reflect()
print(meta.tables['Book'])
Book

通过Class建表

通过“继承”,可以复用父类字段

from sqlalchemy import (
    create_engine,
    inspect,
    Column,
    String,
    Integer,
    Float)

from sqlalchemy.ext.declarative import declarative_base

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

base = declarative_base()

class TemplateTable(object):
    id = Column(Integer, primary_key=True)
    name = Column(String(20))
    age = Column(Integer)

class Customer(TemplateTable, base):
    __tablename__ = 'customer'

class VipCustomer(TemplateTable, base):
    __tablename__ = 'vipcustomer'
    percentoff = Column(Float)

base.metadata.create_all(bind=engine)
inspector = inspect(engine)

for _t in inspector.get_table_names():
    print(_t)
Author
Book
customer
example
vipcustomer

删除表

from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import inspect
from sqlalchemy import Table
from sqlalchemy import Column, Integer, String

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

meta = MetaData()

test = Table('test', meta,
            Column('id', Integer, primary_key=True),
            Column('val', String))

test.create(engine)

inspector = inspect(engine)
print('test' in inspector.get_table_names())
True
test.drop(engine)
inspector = inspect(engine) #需要重新检测数据库
print('test' in inspector.get_table_names())
False

操作Table对象

from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer, String

meta = MetaData()
t = Table('test',meta,
         Column('id', primary_key=True),
         Column('title', String))

# 表名
print(t.name)

# 所有字段
print(t.columns.keys())

# 1个字段
c = t.c.title # 或者c = t.columns.id
print(c.name)

# 通过字段得到
print(c.table)
test
['id', 'title']
title
test

SQL 表达式语言

from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import Integer, String
from sqlalchemy import or_

meta = MetaData()
student = Table('student', meta,
               Column('id', Integer, primary_key=True),
               Column('name', String),
               Column('grade', String))

print(repr(student.c.name=='张三'))
print(str(student.c.name=='张三'))
print(student.c.name=='张三')

student.name = :name_1
student.name = :name_1
print(student.c.id > 3)
print((student.c.id > 3) | (student.c.name != '张三'))
student.id > :id_1
student.id > :id_1 OR student.name != :name_1
print(student.c.name==None)
student.name IS NULL
print(student.c.id + 3)
student.id + :id_1
print(student.c.name + '李四')
student.name || :name_1
print(student.c.name in ('张三', '李四'))
False

insert() 创建insert语句

from sqlalchemy import create_engine
from sqlalchemy import MetaData
from sqlalchemy import Table

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

meta = MetaData(engine)
meta.reflect()

# 得到book表
book = meta.tables['Book']
# 构建Insert SQL
# INSERT INTO book (title, author) values ('三体', '刘慈欣')
ins = book.insert().values(
        title='三体',
        author='刘慈欣')

# 执行插入
conn = engine.connect()
conn.execute(ins)
rs = conn.execute('SELECT * FROM book')
rs.fetchall()
[(1, '三体', '刘慈欣')]
# 插入多条数据
rs = conn.execute(book.insert(), [
    {'title':'狂人日记', 'author':'鲁迅'},
    {'title':'海燕', 'author':'高尔基'}])
rs.rowcount
2

select() 创建select语句

# SELECT * FORM book
select_st = book.select()

rs = conn.execute(select_st)
for _r in rs:
    print(_r)
(1, '三体', '刘慈欣')
(2, '狂人日记', '鲁迅')
(3, '海燕', '高尔基')
# SELECT * FROM book WHERE title = '三体'
select_st = book.select().where(book.c.title=='三体')

rs = conn.execute(select_st)
for _r in rs:
    print(_r)
(1, '三体', '刘慈欣')
# SELECT * FROM book WHERE title = '三体' or  author = '鲁迅'
select_st = book.select().where(or_(book.c.title=='三体', book.c.author=='鲁迅'))

rs = conn.execute(select_st)
for _r in rs:
    print(_r)
(1, '三体', '刘慈欣')
(2, '狂人日记', '鲁迅')
# SELECT * FROM book where id > 1 order by name
select_st = book.select().where(book.c.id >1).order_by(book.c.title)
rs = conn.execute(select_st)
for _r in rs:
    print(_r)
(3, '海燕', '高尔基')
(2, '狂人日记', '鲁迅')

join() 连接表

book = meta.tables['Book']
author = meta.tables['Author']
author.columns.keys()
['id', 'name', 'bio']
# 先插一些数据
conn.execute(author.insert(),
            [{'name':'刘慈欣', 'bio':'中国科幻作家'},
            {'name':'鲁迅', 'bio':'文学家,思想家,教育家'},
             {'name':'高尔基', 'bio':'俄国作家'}])
conn.execute(book.insert(),
            [{'title':'孔乙己', 'author':'鲁迅'},
            {'title':'流浪地球', 'author':'刘慈欣'},
            {'title':'少年闰土', 'author':'鲁迅'},
            {'title':'阿Q正传', 'author':'鲁迅'}])
from sqlalchemy import select

# 创建join对象
join_obj = book.join(author, book.c.author==author.c.name)

# 创建查询对象
select_st = select([book.c.title, book.c.author, author.c.bio]).select_from(join_obj)

rs = conn.execute(select_st)
for _r in rs:
    print(_r)
('三体', '刘慈欣', '中国科幻作家')
('狂人日记', '鲁迅', '文学家,思想家,教育家')
('海燕', '高尔基', '俄国作家')
('孔乙己', '鲁迅', '文学家,思想家,教育家')
('流浪地球', '刘慈欣', '中国科幻作家')
('少年闰土', '鲁迅', '文学家,思想家,教育家')
('阿Q正传', '鲁迅', '文学家,思想家,教育家')

delete()删除行

from sqlalchemy import create_engine
from sqlalchemy import MetaData

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

meta = MetaData(engine)
meta.reflect()

meta.tables.keys()
dict_keys(['Author', 'Book', 'customer', 'example', 'vipcustomer'])
example = meta.tables['example']
conn = engine.connect()
rs = conn.execute(example.select())
rs.fetchall()
[(1, '小花')]
# DELETE FROM example WHERE name = '小花'
delete_st = example.delete().where(example.c.name=='小花')

# 删除
conn.execute(delete_st)

# 检查看是否还有数据
rs = conn.execute(example.select())
print(rs.fetchall())

检查表是否存在

from sqlalchemy import create_engine
from sqlalchemy import inspect
from sqlalchemy import MetaData

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

# 使用inspect检查
inspector = inspect(engine)
for _t in inspector.get_table_names():
    print(_t)

# 使用MetaData检查
meta = MetaData(engine)
meta.reflect()
for _t in meta.tables:
    print(_t)
Author
Book
customer
example
vipcustomer
Author
Book
customer
example
vipcustomer

ORM

添加数据

from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

Model = declarative_base()

class Movie(Model):
    __tablename__ = 'movie'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    director = Column(String)
    time = Column(DateTime, default=datetime.utcnow)

# 建表
Model.metadata.create_all(bind=engine)

# 创建session
Session = sessionmaker() # 工厂函数,返回Session类
Session.configure(bind=engine) #绑定数据库
session = Session()# 得到session对象

data = [
    {'name':'十面埋伏', 'director':'张艺谋'},
    {'name':'无极', 'director':'张凯歌'},
    {'name':'手机', 'director':'冯小刚'}
]

try:
    for item in data:
        row = Movie(name=item['name'], director=item['director'])
        session.add(row)
    session.commit()
except SQLAlchemyError as e:
    print(e)
finally:
    session.close()

更新数据

row = session.query(Movie).filter(Movie.name=='无极').first()
row
<__main__.Movie at 0xd94310>
row.name
'无极'
row.name = '一个馒头引发的血案'
session.commit()

删除数据

mantou = session.query(Movie).filter(Movie.name.like('%馒头%')).first()
mantou.name
'一个馒头引发的血案'
# 删除
session.delete(mantou)

# 查看还有没有
rs = session.query(Movie).all()
for r in rs:
    print(r.name)
十面埋伏
手机

模型关系

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Model = declarative_base()

class User(Model):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    addresses = relationship('Address', back_populates='user')

class Address(Model):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True)
    email = Column(String)
    user_id = Column(Integer, ForeignKey('user.id'))
    user = relationship('User', back_populates='addresses')

u1 = User()
a1 = Address()
print(u1.addresses)
print(a1.user)

u1.addresses.append(a1)
print(u1.addresses)
print(a1.user)
[]
None
[<__main__.Address object at 0x03C000F0>]
<__main__.User object at 0x03C15350>

自关联

import json

from sqlalchemy import Column, Integer, String, Table, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Model = declarative_base()

node_assosiation = Table('node_assos', Model.metadata,
                        Column('left', Integer, ForeignKey('node.id')),
                        Column('right', Integer, ForeignKey('node.id')))

class Node(Model):
    __tablename__ = 'node'
    id = Column(Integer, primary_key=True)
    label = Column(String)
    friends = relationship('Node',
                           secondary=node_assosiation,
                           primaryjoin=id==node_assosiation.c.left,
                           secondaryjoin=id==node_assosiation.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"
    ]
  }
]

基本查询

from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy import or_, desc
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, session

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

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

print('------>按时间排序')
for r in  session.query(Movie).order_by(Movie.time).all():
    print(r.id, r.name, r.director, r.time)
------>按时间排序
1 十面埋伏 张艺谋 2019-07-23 07:41:38.315399
2 一个馒头引发的血案 张凯歌 2019-07-23 07:41:38.317395
3 手机 冯小刚 2019-07-23 07:41:38.317395
print('------>按id排序-倒序')
for r in  session.query(Movie).order_by(desc(Movie.time)).all():
    print(r.id, r.name, r.director, r.time)
------>按id排序-倒序
2 一个馒头引发的血案 张凯歌 2019-07-23 07:41:38.317395
3 手机 冯小刚 2019-07-23 07:41:38.317395
1 十面埋伏 张艺谋 2019-07-23 07:41:38.315399
print('------>==过滤')
for r in  session.query(Movie).filter(Movie.director=='张凯歌').all():
    print(r.id, r.name, r.director, r.time)
------>==过滤
2 一个馒头引发的血案 张凯歌 2019-07-23 07:41:38.317395
print('------>!=过滤')
for r in  session.query(Movie).filter(Movie.director!='张凯歌').all():
    print(r.id, r.name, r.director, r.time)
------>!=过滤
1 十面埋伏 张艺谋 2019-07-23 07:41:38.315399
3 手机 冯小刚 2019-07-23 07:41:38.317395
print('------>IN 过滤')
for r in  session.query(Movie).filter(Movie.director.in_(['张凯歌', '张艺谋'])).all(): # 注意,是in_
    print(r.id, r.name, r.director, r.time)
------>IN 过滤
1 十面埋伏 张艺谋 2019-07-23 07:41:38.315399
2 一个馒头引发的血案 张凯歌 2019-07-23 07:41:38.317395
print('------>NOT IN过滤')
for r in  session.query(Movie).filter(~Movie.director.in_(['张凯歌', '张艺谋'])).all():
    print(r.id, r.name, r.director, r.time)
------>NOT IN过滤
3 手机 冯小刚 2019-07-23 07:41:38.317395
print('------>AND')
for r in  session.query(Movie).filter(Movie.name.like('%手机%'), Movie.director=='冯小刚').all():
    print(r.id, r.name, r.director, r.time)
------>AND
3 手机 冯小刚 2019-07-23 07:41:38.317395
print('------>OR')
for r in  session.query(Movie).filter(or_(Movie.name.like('%埋伏%'), Movie.director=='冯小刚')).all():
    print(r.id, r.name, r.director, r.time)
------>OR
1 十面埋伏 张艺谋 2019-07-23 07:41:38.315399
3 手机 冯小刚 2019-07-23 07:41:38.317395
print('------>NULL')
for r in  session.query(Movie).filter(Movie.time == None).all():
    print(r.id, r.name, r.director, r.time)
------>NULL
print('------>NOT NULL')
for r in  session.query(Movie).filter(Movie.time != None).all():
    print(r.id, r.name, r.director, r.time)
------>NOT NULL
1 十面埋伏 张艺谋 2019-07-23 07:41:38.315399
2 一个馒头引发的血案 张凯歌 2019-07-23 07:41:38.317395
3 手机 冯小刚 2019-07-23 07:41:38.317395
print('------>LIKE')
for r in  session.query(Movie).filter(Movie.name.like('%埋伏%')).all():
    print(r.id, r.name, r.director, r.time)
------>LIKE
1 十面埋伏 张艺谋 2019-07-23 07:41:38.315399

Table 映射成 Class

from sqlalchemy import MetaData, create_engine

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)
meta = MetaData(bind=engine)
meta.reflect()
book = meta.tables['Book']
author = meta.tables['Author']

print(book.columns.keys())
print(author.columns.keys())
['id', 'title', 'author']
['id', 'name', 'bio']
from sqlalchemy.orm import mapper, relationship, sessionmaker

class Book:
    def __init__(self,title, author):
        self.title, self.author = title, author

class Author:
    def __init__(self,name, bio):
        self.name, self.bio = name, bio

mapper(Author, author) # 使用properties参数,传入一个字段名和字段值的键值对,可以新增字段
mapper(Book, book)

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

a = Author('老舍', '中国近代著名作家')
b = Book('骆驼祥子', '老舍')
session.add(a)
session.add(b)
try:
    session.commit()
    b = session.query(Book).filter(Book.title=='骆驼祥子').first()
    print(b.title, b.author)
finally:
    session.close()
骆驼祥子 老舍

从数据库中获取Table,动态转成Class

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker, mapper

db_uri = 'sqlite:///db.sqlite'
engine = create_engine(db_uri)

meta = MetaData(engine)
meta.reflect()

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

def get_table(name):
    if name not in meta.tables: # 这里也可以增加逻辑 :如果没有表,则自动创建
        return None
    table = meta.tables[name]
    cls = type(name, (), {}) # 创建类
    mapper(cls, table)
    return cls

Book = get_table('Book')

for b in session.query(Book).all():
    print(b.title)
三体
狂人日记
海燕
孔乙己
流浪地球
少年闰土
阿Q正传
骆驼祥子

ORM中的JOIN

from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

db_uri = 'sqlite:///'
engine = create_engine(db_uri)

Session = sessionmaker(bind=engine)
session = Session()

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

base.metadata.create_all(engine) # 建表

user1 = User(name='张三')
email1 = Address(email='abcd1@xxx.com')
email2 = Address(email='abcd2@xxx.com')
user1.addresses.extend([email1, email2])

session.add(user1)
session.add_all([email1, email2])
session.commit()

query = session.query(User, Address).join(User)
for _u, _a in query.all():
    print(_u.name, _a.email)
张三 abcd1@xxx.com
张三 abcd2@xxx.com

JOIN后GROUP求COUNT

user2 = User(name='李四')
email3 = Address(email='lisi1@xxx.com')
email4 = Address(email='lisi2@xxx.com')
session.add(user2)
session.add_all([email3, email4])
session.commit()
from sqlalchemy import func
query = session.query(User, func.count(Address.id)).join(User).group_by(User.id)
for _u, _c in query.all():
    print(_u.name, _c)
张三 2

关闭数据库连接

from sqlalchemy import event

@event.listens_for(engine, 'engine_disposed') # 监听引擎关闭事件
def receive_engine_disposed(engine):
    print("engine dispose")

engine.dispose()
engine dispose

参考

Leave a Reply

Your email address will not be published. Required fields are marked *