编程►python

python中使用SQLAlchemy(二)

关于SQLAlchemy使用建表、插入和查询内容，点击python中使用SQLAlchemy查看。
继续以前面建立的student表作为示例进行后面的操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
mysql> desc student;
+---------+--------------+------+-----+---------+----------------+
| Field   | Type         | Null | Key | Default | Extra          |
+---------+--------------+------+-----+---------+----------------+
| id      | int(11)      | NO   | PRI | NULL    | auto_increment |
| name    | varchar(100) | YES  |     | NULL    |                |
| age     | int(11)      | YES  |     | NULL    |                |
| address | varchar(100) | YES  |     | NULL    |                |
+---------+--------------+------+-----+---------+----------------+
4 rows in set (0.15 sec)
mysql> select * from student;
+------+-------+------+---------+
| id   | name  | age  | address |
+------+-------+------+---------+
| 1001 | Alice |   25 | anhui   |
| 1002 | Bob   |   69 | beijing |
| 1003 | Cerry |   14 | jiangsu |
+------+-------+------+---------+
3 rows in set (0.07 sec)

更新

更新时，首先查询需要更新的数据，然后直接更新对应字段即可。

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, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine("mysql+pymysql://root:123456@localhost/test")
DBsession = sessionmaker(bind=engine)
session = DBsession()
Base = declarative_base()

class Student(Base):
    __tablename__ = 'student'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    age = Column(Integer)
    address = Column(String(100))

student1 = session.query(Student).filter(Student.name == 'Alice').one()
print("更新前age：{0}".format(student1.age))
student1.age = '38'
session.commit()
print("更新后age：{0}".format(student1.age))
session.close()

输出结果

1
2
更新前age：25
更新后age：38

删除

删除时，首先查询出需要删除的数据，然后直接调用delete()方法直接删除即可。

1
2
3
student2 = session.query(Student).filter(Student.name == 'Alice').delete()
session.commit()
session.close()

数据库结果确认

1
2
3
4
5
6
7
8
mysql> select * from student;
+------+-------+------+---------+
| id   | name  | age  | address |
+------+-------+------+---------+
| 1002 | Bob   |   69 | beijing |
| 1003 | Cerry |   14 | jiangsu |
+------+-------+------+---------+
2 rows in set (0.00 sec)

在进行分组及排序操作前，先插入几条数据

1
2
3
4
5
6
7
student1 = Student(id=1001, name='Alice', age=22, address='beijing')
student2 = Student(id=1004, name='Dany', age=14, address='beijing')
student3 = Student(id=1005, name='Ever', age=97, address='beijing')
student4 = Student(id=1006, name='For', age=50, address='beijing')
session.add_all([student1, student2, student3, student4])
session.commit()
session.close()

数据库结果确认

1
2
3
4
5
6
7
8
9
10
11
12
mysql> select * from student;
+------+-------+------+---------+
| id   | name  | age  | address |
+------+-------+------+---------+
| 1001 | Alice |   22 | beijing |
| 1002 | Bob   |   69 | beijing |
| 1003 | Cerry |   14 | jiangsu |
| 1004 | Dany  |   14 | beijing |
| 1005 | Ever  |   97 | beijing |
| 1006 | For   |   50 | beijing |
+------+-------+------+---------+
6 rows in set (0.00 sec)

统计

统计方法为count()

1
2
count_student = session.query(Student).filter(Student.name.like('%e%')).count()
print("姓名中带字母e的人{0}个".format(count_student))

输出结果

1
姓名中带字母e的人3个

分组

分组方法为group_by()

1
2
3
group_student = session.query(Student).group_by(Student.address).all()
for i in group_student:
    print(i.id, i.name, i.age, i.address)

输出结果

1
2
1001 Alice 22 beijing
1003 Cerry 14 jiangsu

排序

排序方法为order_by()，默认为升序，反序在order_by里面使用desc()方法。

1
2
3
4
5
6
7
8
9
order_student = session.query(Student).filter(Student.age > 30).order_by(Student.age).all()
print("默认排序输出")
for x in order_student:
    print(x.id, x.name, x.age, x.address)

orderdesc_student = session.query(Student).filter(Student.age > 30).order_by(Student.age.desc()).all()
print("反序输出")
for y in orderdesc_student:
    print(y.id, y.name, y.age, y.address)

输出结果

1
2
3
4
5
6
7
8
默认排序输出
1006 For 50 beijing
1002 Bob 69 beijing
1005 Ever 97 beijing
反序输出
1005 Ever 97 beijing
1002 Bob 69 beijing
1006 For 50 beijing

总结

基本步骤

  1. 创建引擎engine

    1
    engine = create_engine("mysql+pymysql://root:123456@localhost/test")

  2. 创建session

    1
    2
    DBsession = sessionmaker(bind=engine)
    session = DBsession()

  3. 定义模型

    1
    2
    3
    4
    5
    6
    7
    8
    Base = declarative_base()

    class Student(Base):    #Student类继承自Base类，
        __tablename__ = 'student'    #对应数据库表名
        id = Column(Integer, primary_key=True)    #对应数据库各字段
        name = Column(String(100))
        age = Column(Integer)
        address = Column(String(100))

接下来基于前面创建的session和Student类及属性进行增删改查的操作。

常见查询总结

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
#简单查询
print(session.query(Student).all())
print(session.query(Student.id, Student.name).all())
print(session.query(Student, Student.name).all())
#带条件查询
print(session.query(Student).filter_by(name='Bob').all())
print(session.query(Student).filter(Student.name == 'Bob').all())
print(session.query(Student).filter(Student.name.like('%e%')).all())
#多条件查询
from sqlalchemy import and_, or_
print(session.query(Student).filter(and_(Student.age > 20, Student.name.like('%e%'))).all())
print(session.query(Student).filter(or_(Student.age > 20, Student.name.like('%e%'))).all())
#sql过滤
print(session.query(Student).filter("id=:id").params(id=1001).all())
#关联查询
print(session.query(User, Address).filter(User.id == Address.user_id).all())
print(session.query(User).join(User.addresses).all())
print(session.query(User).outerjoin(User.addresses).all())
#聚合查询
print(session.query(User.name, func.count('*').label("user_count")).group_by(User.name).all())
print(session.query(User.name, func.sum(User.id).label("user_id_sum")).group_by(User.name).all())
#子查询
 stmt = session.query(Address.user_id, func.count('*').label("address_count")).group_by(Address.user_id).subquery()
print(session.query(User, stmt.c.address_count).outerjoin((stmt, User.id == stmt.c.user_id)).order_by(User.id).all())
#exists
print(session.query(User).filter(exists().where(Address.user_id == User.id)))
print(session.query(User).filter(User.addresses.any()))

限制返回字段查询

1
2
3
person = session.query(Person.name, Person.created_at,                     
             Person.updated_at).filter_by(name="zhongwei").order_by(            
             Person.created_at).first()

记录总数查询

1
2
3
4
5
6
7
8
9
10
from sqlalchemy import func

session.query(func.count(User.id))

session.query(func.count(User.id)).\
        group_by(User.name)

from sqlalchemy import distinct

session.query(func.count(distinct(User.name)))

Recommended Posts