Python 编程之sqlalchemy ORM

ORM介绍

orm英文全称object relational mapping,就是对象映射关系程序，简单来说我们类似python这种面向对象的程序来说一切皆对象，但是我们使用的数据库却都是关系型的，为了保证一致的使用习惯，通过orm将编程语言的对象模型和数据库的关系模型建立映射关系，这样我们在使用编程语言对数据库进行操作的时候可以直接使用编程语言的对象模型进行操作就可以了，而不用直接使用sql语言。

orm的优点：

隐藏了数据访问细节，“封闭”的通用数据库交互，ORM的核心。他使得我们的通用数据库交互变得简单易行，并且完全不用考虑该死的SQL语句。快速开发，由此而来。
ORM使我们构造固化数据结构变得简单易行。

缺点：

无可避免的，自动化意味着映射和关联管理，代价是牺牲性能（早期，这是所有不喜欢ORM人的共同点）。现在的各种ORM框架都在尝试使用各种方法来减轻这块（LazyLoad，Cache），效果还是很显著的。

sqlalchemy基本使用

在没有orm之前，我们创建一个表是这样的：

CREATE TABLE user (
    id INTEGER NOT NULL AUTO_INCREMENT,
    name VARCHAR(32),
    password VARCHAR(64),
    PRIMARY KEY (id)
)

这只是最简单的sql表，如果再加上外键关联什么的，一般程序员是很难记住那些sql语句的，于是有了orm,实现上面同样的功能，代码如下：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,DateTime,Enum,ForeignKey,UniqueConstraint,ForeignKeyConstraint,Index

engine = create_engine("mysql+pymysql://root:[email protected]:3306/test",
                       encoding='utf-8', echo=True)

Base = declarative_base() #生成orm基类

class User(Base):   #必须继承declaraive_base得到的那个基类
    __tablename__ = 'user'  #必须要有__tablename__来指出这个类对应什么表，这个表可以暂时在库中不存在，SQLAlchemy会帮我们创建这个表
    id = Column(Integer, primary_key=True)  #Column类定义一个字段
    name = Column(String(32),nullable=False,index=True) #nullable=False 表示此列不允许为空。index=True表示为此列创建一个索引
    password = Column(String(64),nullable=False)

Base.metadata.create_all(engine)  #创建表结构

在进行数据操作之前，先生成对应的会话实例，如下所示：

from sqlalchemy.orm import sessionmaker

Session_class = sessionmaker(bind=engine)  #创建与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
Session = Session_class()  #生成session实例

insert

user_obj = User(name="tom", password="tom123")  # 生成你要创建的数据对象
print(user_obj.name, user_obj.id)  # 此时还没创建对象呢，不信你打印一下id发现还是None

Session.add(user_obj)  # 把要创建的数据对象添加到这个session里， 一会统一创建
Session.add_all([
    User(name="jack", password="jack123"),
    User(name="lily", password="lily123"),
    User(name="dave", password="dave123")
])
print(user_obj.name, user_obj.id)  # 此时也依然还没创建

Session.commit()  # 现此才统一提交，创建数据

select

#查所有，取所有字段
res = Session.query(User).all()
for row in res:
    print(row.id,row.name,row.password)

#查所有，取指定字段
res=Session.query(User.name).order_by(User.id).all()
for row in res:
    print(row.name)

res=Session.query(User.name).first()
print(res.name)

#过滤查
res=Session.query(User).filter(User.id > 1,User.id < 4)    #逗号分隔,默认为and。相当于user.id > 1 and user.id << 4的效果
for row in res:
    print(row.id,row.name)

update

Session.query(User).filter(User.id > 0).update({'name':'哇哈哈'})
Session.query(User).filter(User.id > 0).update({'name':User.name+'_SB'},synchronize_session=False)
Session.query(User).filter(User.id > 0).update({'id':User.id*100},synchronize_session='evaluate')

Session.commit()

delete

Session.query(User).filter(User.id > 3).delete()
Session.commit()

回滚：

my_user = Session.query(User).filter_by(id=1).first()
my_user.name = "Jack"
 
 
fake_user = User(name='Rain', password='12345')
Session.add(fake_user)
 
print(Session.query(User).filter(User.name.in_(['Jack','rain'])).all() )  #这时看session里有你刚添加和修改的数据
 
Session.rollback() #此时你rollback一下
 
print(Session.query(User).filter(User.name.in_(['Jack','rain'])).all() ) #再查就发现刚才添加的数据没有了。
 
# Session
# Session.commit()

其他查询相关

1. 准备表和数据

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

engine = create_engine("mysql+pymysql://root:[email protected]:3306/test",
                       encoding='utf-8', echo=True)

Base=declarative_base()


#多对一:假设多个员工可以属于一个部门,而多个部门不能有同一个员工(只有创建公司才把员工当骆驼用,一个员工身兼数职)
class Dept(Base):
    __tablename__='dept'
    id=Column(Integer,primary_key=True,autoincrement=True)
    dname=Column(String(64),nullable=False,index=True)

class Emp(Base):
    __tablename__='emp'
    id=Column(Integer,primary_key=True,autoincrement=True)
    ename=Column(String(32),nullable=False,index=True)
    dept_id=Column(Integer,ForeignKey('dept.id'))

def init_db():
    Base.metadata.create_all(engine)

def drop_db():
    Base.metadata.drop_all(engine)

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

# 准备数据
session.add_all([
    Dept(dname='技术'),
    Dept(dname='销售'),
    Dept(dname='运营'),
    Dept(dname='人事'),
])

session.add_all([
    Emp(ename='张三',dept_id=1),
    Emp(ename='李四',dept_id=1),
    Emp(ename='王五',dept_id=1),
    Emp(ename='露西',dept_id=2),
    Emp(ename='赵九',dept_id=3),
    Emp(ename='张二丫',dept_id=4),
    Emp(ename='李坦克',dept_id=2),
    Emp(ename='王大炮',dept_id=4),
    Emp(ename='牛榴弹',dept_id=3)
])

session.commit()

View Code

2. 条件、通配符、limit、排序、分组、连表、组合

#一、条件
res = session.query(Emp).filter_by(ename='张三').all()    #filter_by只能传参数:什么等于什么
for row in res:
    print(row.id,row.ename)

res=session.query(Emp).filter(Emp.id>0,Emp.ename == '张三').all() #filter内传的是表达式,逗号分隔,默认为and,
res=session.query(Emp).filter(Emp.id.between(1,3),Emp.ename == '张三').all()  #select * from emp where ename='张三' and id between 1 and 3;
res=session.query(Emp).filter(Emp.id.in_([1,3,99,101]),Emp.ename == '张三').all()    #select * from emp where ename='张三' and id in(1,3,99,101);
res=session.query(Emp).filter(~Emp.id.in_([1,3,99,101]),Emp.ename == '张三') #~代表取反,转换成sql就是关键字not

from sqlalchemy import and_,or_
res=session.query(Emp).filter(and_(Emp.id > 0,Emp.ename=='张三')).all()  #select * from emp where id > 0 and ename='张三';
res=session.query(Emp).filter(or_(Emp.id < 2,Emp.ename=='李四')).all()  #select * from emp where id < 2 or ename='李四';
res=session.query(Emp).filter(      #select * from emp where dept_id = 3 or ename != '' or (id > 1 and ename = '李四');
    or_(
        Emp.dept_id == 3,
        and_(Emp.id > 1,Emp.ename=='李四'),
        Emp.ename != ''
    )
).all()

#二、通配符
res=session.query(Emp).filter(Emp.ename.like('%坦_')).all()  #select * from emp where ename like '%坦%';
res=session.query(Emp).filter(~Emp.ename.like('%坦_')).all() #select * from emp where ename not like '%坦%';

#三、limit
res=session.query(Emp)[0:5:2]

#四、排序
res=session.query(Emp).order_by(Emp.dept_id.desc()).all()   #select * from emp order by dept_id desc;
res=session.query(Emp).order_by(Emp.dept_id.desc(),Emp.id.asc()).all()  #select * from emp order by dept_id desc,id asc;

#五、分组
from sqlalchemy.sql import func

res=session.query(Emp.dept_id).group_by(Emp.dept_id).all()  #select dept_id from emp group by dept_id;
res=session.query(  #select dept_id,max(dept_id),min(dept_id),sum(dept_id),avg(dept_id),count(dept_id) from emp group by dept_id;
    func.max(Emp.dept_id),
    func.min(Emp.dept_id),
    func.sum(Emp.dept_id),
    func.avg(Emp.dept_id),
    func.count(Emp.dept_id),
).group_by(Emp.dept_id).all()


res=session.query(  #select dept_id,count(1) from emp group by dept_id having count(1) > 2;
    Emp.dept_id,
    func.count(1),
).group_by(Emp.dept_id).having(func.count(1) > 2).all()

#六、连接
#笛卡尔积
res=session.query(Emp,Dept).all() #select * from emp,dep;

#where条件
res=session.query(Emp,Dept).filter(Emp.dept_id==Dept.id).all()  #select * from emp,dept where emp.dept_id = dept.id;
for row in res:
    emp_tb=row[0]
    dep_tb=row[1]
    print(emp_tb.id,emp_tb.ename,dep_tb.id,dep_tb.dname)

#内连接
res=session.query(Emp).join(Dept)
#join默认为内连接,SQLAlchemy会自动帮我们通过foreign key字段去找关联关系
#但是上述查询的结果均为Emp表的字段,这样链表还有毛线意义,于是我们修改为
res=session.query(Emp.id,Emp.ename,Emp.dept_id,Dept.dname).join(Dept).all()

#左连接:isouter=True
res=session.query(Emp.id,Emp.ename,Emp.dept_id,Dept.dname).join(Dept,isouter=True).all()

#右连接:同左连接,只是把两个表的位置换一下


#七、组合
q1=session.query(Emp.id,Emp.ename).filter(Emp.id > 0,Emp.id < 5)
q2=session.query(Emp.id,Emp.ename).filter(
    or_(
        Emp.ename.like('%坦%'),
        Emp.ename.like('%大%'),
    )
)
res1=q1.union(q2) #组合+去重
res2=q1.union_all(q2) #组合,不去重

print([i.ename for i in q1.all()]) #['张三', '李四', '王五', '露西']
print([i.ename for i in q2.all()]) #['李坦克', '王大炮']
print([i.ename for i in res1.all()]) #['张三', '李四', '王五', '露西', '李坦克', '王大炮']
print([i.ename for i in res2.all()]) #['张三', '李四', '王五', '露西', '李坦克', '王大炮']

View Code

3. 子查询

有三种形式的子查询，注意：子查询的sql必须用括号包起来,尤其在形式三中需要注意这一点

#示例:查出id大于2的员工,当做子查询的表使用

#原生SQL:
# select * from (select * from emp where id > 2);

#ORM：
res=session.query(
    session.query(Emp).filter(Emp.id > 2).subquery()
).all()

for row in res:
    print(row.id,row.ename,row.dept_id)

形式一:子查询当做一张表来用,调用subquery()

#示例:#查出销售部门的员工姓名

#原生SQL:
# select ename from emp where dept_id in (select id from dept where dname='销售');

#ORM：
res=session.query(Emp.ename).filter(Emp.dept_id.in_(
    session.query(Dept.id).filter_by(dname='销售'), #传的是参数
    # session.query(Dep.id).filter(Dep.dname=='销售') #传的是表达式
)).all()

形式二:子查询当做in的范围用,调用in_

#示例:查询所有的员工姓名与部门名

#原生SQL:
# select ename as 员工姓名,(select dname from dept where id = emp.dept_id) as 部门名 from emp;

#ORM：
sub_sql=session.query(Dept.dname).filter(Dept.id==Emp.dept_id) #SELECT dept.dname FROM dept, emp WHERE dept.id = emp.dept_id
sub_sql.as_scalar() #as_scalar的功能就是把上面的sub_sql加上了括号

res=session.query(Emp.ename,sub_sql.as_scalar()).all()

形式三:子查询当做select后的字段,调用as_scalar()

外键关联

先创建一个addresses表，跟user表关联

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship


class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    email_address = Column(String(32), nullable=False)
    user_id = Column(Integer, ForeignKey('user.id'))

    user = relationship("User", backref="addresses")  # 允许你在user表里通过backref字段反向查出所有它在addresses表里的关联项

    def __repr__(self):
        return "<Address(email_address='%s')>" % self.email_address

Base.metadata.create_all(engine)

#准备数据
Session.add_all([
    Address(id=10,email_address='[email protected]',user_id=2),
    Address(id=11,email_address='[email protected]',user_id=1),
    Address(id=12,email_address='[email protected]',user_id=3)
])

Session.commit()

标准连表查询

# 示例:查询员工id、姓名及其邮件名
res=Session.query(User.id,User.name,Address.email_address).join(Address) #迭代器
for row in res:
    print(row[0],row[1],row[2]) #等同于print(row.id,row.name,row.email_address)

基于relationship的正查、反查

#SQLAlchemy的relationship在内部帮我们做好表的链接

#正向查
res = Session.query(Address)
for row in res:
    print(row.id,row.email_address,row.user.name)

#反向查
res=Session.query(User)
for row in res:
    # print(row.dname,row.xxoo)
    print(row.name,[r.email_address for r in row.addresses])

表创建好后，可以这样反查试试

obj = Session.query(User).first()
for i in obj.addresses:  # 通过user对象反查关联的addresses记录
    print(i)

addr_obj = Session.query(Address).first()
print(addr_obj.user.name)  # 在addr_obj里直接查关联的user表

创建关联对象

obj = Session.query(User).filter(User.name == 'tom').all()[0]
print(obj.addresses)

obj.addresses = [Address(email_address="[email protected]"),  # 添加关联对象
                 Address(email_address="[email protected]")]

Session.commit()

常用查询语法

Common Filter Operators

Here’s a rundown of some of the most common operators used in filter():

equals:
```
query.filter(User.name == 'ed')
      
```
not equals:
```
query.filter(User.name != 'ed')
      
```
LIKE:

query.filter(User.name.like('%ed%'))
IN:
NOT IN:

query.filter(~User.name.in_(['ed', 'wendy', 'jack']))
IS NULL:
IS NOT NULL:
AND:

2.1. ObjectRelationalTutorial 17

query.filter(User.name.in_(['ed', 'wendy', 'jack']))

# works with query objects too:

query.filter(User.name.in_( session.query(User.name).filter(User.name.like('%ed%'))

))

query.filter(User.name == None)

# alternatively, if pep8/linters are a concern

query.filter(User.name.is_(None))

query.filter(User.name != None)

# alternatively, if pep8/linters are a concern

query.filter(User.name.isnot(None))

SQLAlchemy Documentation, Release 1.1.0b1

# use and_()

from sqlalchemy import and_

query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))

# or send multiple expressions to .filter()

query.filter(User.name == 'ed', User.fullname == 'Ed Jones')

# or chain multiple filter()/filter_by() calls

query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')

Note: Makesureyouuseand_()andnotthePythonandoperator! • OR:

Note: Makesureyouuseor_()andnotthePythonoroperator! • MATCH:

query.filter(User.name.match('wendy'))

Note: match() uses a database-specific MATCH or CONTAINS f

多外键关联

下表中，Customer表有2个字段都关联了Address表

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

engine = create_engine("mysql+pymysql://root:[email protected]:3306/test",
                       encoding='utf-8', echo=True)
Base = declarative_base()


class Customer(Base):
    __tablename__ = 'customer'
    id = Column(Integer, primary_key=True)
    name = Column(String(20))

    billing_address_id = Column(Integer, ForeignKey("address.id"))
    shipping_address_id = Column(Integer, ForeignKey("address.id"))

    billing_address = relationship("Address")
    shipping_address = relationship("Address")


class Address(Base):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True)
    street = Column(String(32))
    city = Column(String(32))
    state = Column(String(10))

Base.metadata.create_all(engine)

创建表结构是没有问题的，但你Address表中插入数据时会报下面的错

sqlalchemy.exc.AmbiguousForeignKeysError: Could not determine join
condition between parent/child tables on relationship
Customer.billing_address - there are multiple foreign key
paths linking the tables.  Specify the 'foreign_keys' argument,
providing a list of those columns which should be
counted as containing a foreign key reference to the parent table.

解决办法如下：

class Customer(Base):
    __tablename__ = 'customer'
    id = Column(Integer, primary_key=True)
    name = Column(String(20))

    billing_address_id = Column(Integer, ForeignKey("address.id"))
    shipping_address_id = Column(Integer, ForeignKey("address.id"))

    billing_address = relationship("Address", foreign_keys=[billing_address_id])
    shipping_address = relationship("Address", foreign_keys=[shipping_address_id])

这样sqlachemy就能分清哪个外键是对应哪个字段了

多对多关系

现在来设计一个能描述“图书”与“作者”的关系的表结构，需求是

一本书可以有好几个作者一起出版
一个作者可以写好几本书

此时你会发现，用之前学的外键好像没办法实现上面的需求了，此时，我们可以再搞出一张中间表，就可以了。用orm表示如下：

#一本书可以有多个作者，一个作者又可以出版多本书

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

engine = create_engine("mysql+pymysql://root:[email protected]:3306/test",
                       encoding='utf-8', echo=True)
Base = declarative_base()

book_m2m_author = Table('book_m2m_author', Base.metadata,
                        Column('book_id',Integer,ForeignKey('books.id')),
                        Column('author_id',Integer,ForeignKey('authors.id')),
                        )

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer,primary_key=True)
    name = Column(String(64))
    pub_date = Column(DATE)
    authors = relationship('Author',secondary=book_m2m_author,backref='books')

    def __repr__(self):
        return self.name

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))

    def __repr__(self):
        return self.name

Base.metadata.create_all(engine)

接下来创建几本书和作者

Session_class = sessionmaker(bind=engine)  # 创建与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
s = Session_class()  # 生成session实例

b1 = Book(name="跟Alex学Python")
b2 = Book(name="跟Alex学把妹")
b3 = Book(name="跟Alex学装逼")
b4 = Book(name="跟Alex学开车")

a1 = Author(name="Alex")
a2 = Author(name="Jack")
a3 = Author(name="Rain")

b1.authors = [a1, a2]
b2.authors = [a1, a2, a3]

s.add_all([b1, b2, b3, b4, a1, a2, a3])

s.commit()

此时，手动连上mysql，分别查看这3张表，你会发现，book_m2m_author中自动创建了多条纪录用来连接book和author表

mysql> select * from books;
+----+------------------+----------+
| id | name             | pub_date |
+----+------------------+----------+
|  1 | 跟Alex学Python   | NULL     |
|  2 | 跟Alex学把妹     | NULL     |
|  3 | 跟Alex学装逼     | NULL     |
|  4 | 跟Alex学开车     | NULL     |
+----+------------------+----------+
4 rows in set (0.00 sec)

mysql> select * from authors;
+----+------+
| id | name |
+----+------+
|  1 | Alex |
|  2 | Rain |
|  3 | Jack |
+----+------+
3 rows in set (0.00 sec)

mysql> select * from book_m2m_author;
+---------+-----------+
| book_id | author_id |
+---------+-----------+
|       1 |         3 |
|       2 |         3 |
|       2 |         2 |
|       1 |         1 |
|       2 |         1 |
+---------+-----------+
5 rows in set (0.03 sec)

此时，我们去用orm查一下数据

print('--------通过书表查关联的作者---------')

book_obj = s.query(Book).filter_by(name="跟Alex学Python").first()
print(book_obj.name, book_obj.authors)

print('--------通过作者表查关联的书---------')
author_obj = s.query(Author).filter_by(name="Alex").first()
print(author_obj.name, author_obj.books)
s.commit()

输出如下：

--------通过书表查关联的作者---------
跟Alex学Python [Alex, Jack]
--------通过作者表查关联的书---------
Alex [跟Alex学Python, 跟Alex学把妹]

多对多删除

删除数据时不用管boo_m2m_authors　， sqlalchemy会自动帮你把对应的数据删除

通过书删除作者

author_obj = s.query(Author).filter_by(name="Jack").first()

book_obj = s.query(Book).filter_by(name="跟Alex学把妹").first()

book_obj.authors.remove(author_obj)  # 从一本书里删除一个作者
s.commit()

直接删除作者

author_obj =s.query(Author).filter_by(name="Alex").first()
# print(author_obj.name , author_obj.books)
s.delete(author_obj)
s.commit()

处理中文

sqlalchemy设置编码字符集一定要在数据库访问的URL上增加charset=utf8，否则数据库的连接就不是utf8的编码格式

eng = create_engine('mysql://root:[email protected]:3306/test2?charset=utf8',echo=True)

转载于:https://www.cnblogs.com/cyfiy/p/9341239.html

Python 编程之sqlalchemy ORM

sqlalchemy基本使用

继续阅读

《Linux命令行与Shell脚本编程大全第2版.布卢姆》pdf

MySQL的4种隔离级别？出现问题

XX系统实施过程问题总结

无组件上传图片到数据库中，最完整解决方案

【MySQL数据库】数据库索引事务1.索引2.事务

neo4j之cypher使用文档

Cloud Studio初体验

使用 ctypes 进行 Python 和 C 的混合编程

【python】【数据处理】画多维数据分布图

NOSQL安全攻击

mybatis_入门程序Mybatis入门

登录plsql 报错 the account is locked --用户被锁

SequoiaDB巨杉数据库C++驱动概述

【python】netconf协议对接管理设备

「Python 网络自动化」NETCONF —— Python 使用 NETCONF 管理配置 H3C 网络设备

在python中创建excel并写入