SQLAlchmy模块详解

之前写过一篇博客介绍过sqlalchemy的基本用法，本篇博客主要介绍除增删改查以外SQLAlchemy对数据库表的操作，主要内容有单表操作、一对多操作、多对多操作。

一、单表操作

单表操作的增删改查在上篇博客中已经详细介绍过，这里不再详细介绍，今天主要对数据库查询在详细介绍一下，下面我们先创建表并插入数据。


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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

 

from sqlalchemy import and_, or_

from sqlalchemy import create_engine

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index

from sqlalchemy.orm import sessionmaker, relationship

engine = create_engine("mysql+pymysql://root:[email protected]:3306/mydata", max_overflow=5)

 

Base = declarative_base()

 

class Group(Base):

    __tablename__=‘group‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    caption = Column(String(32))

 

class User(Base):

    __tablename__=‘user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    username = Column(String(32))

    group_id = Column(Integer, ForeignKey(‘group.nid‘))

    group = relationship("Group",backref=‘uuu‘)  #跟Group表建立关系，方便查询,常和ForeignKey在一起使用

 

def init_table():

    """

    创建表，调用Base类的子类

    :return:

    """

    Base.metadata.create_all(engine)

 

def drop_table():

    Base.metadata.drop_all(engine)

 

init_table()

Session = sessionmaker(bind=engine)

session = Session()

 

# 单表操作：

session.add(Group(caption=‘dba‘))   #往组里添加数据

session.add(Group(caption=‘dddd‘))

session.commit()

 

session.add_all([

    User(username=‘jack1‘,group_id=1),

    User(username=‘jack2‘,group_id=1),

    User(username=‘jack1‘,group_id=2),

    User(username=‘jack1‘,group_id=1),

    User(username=‘jack2‘,group_id=1),

])

session.commit()

1，条件查询


1

2

3

4

5

6

7

8

9

#查询用户jack1的nid,filter和filter_by两种书写方式

ret1 = session.query(User.nid).filter(User.username==‘jack1‘).all()

print(ret1)

ret2 = session.query(User.nid).filter_by(username=‘jack1‘).all()

print(ret2)

#结果：

[(1,), (3,), (4,)]

[(1,), (3,), (4,)]

1

2

3

4

5

6

7

8

9

10

11

12

13

#查询用户nid大于1并且username等于jack2的nid

ret1 = session.query(User.nid).filter(User.nid >1,User.username==‘jack2‘).all()

print(ret1)

#结果：

[(2,), (5,)]

#查询nid在1和3之间username等于jack1的所有信息

ret2=session.query(User.nid,User.username).filter(User.nid.between(1,3),User.username==‘jack1‘).all()

print(ret2)

#结果：

[(1, ‘jack1‘), (3, ‘jack1‘)]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

#查询用户nid在1,3,4这个列表里的用户信息

ret=session.query(User.nid,User.username).filter(User.nid.in_([1,3,4])).all()

print(ret)

  

#结果：

[(1, ‘jack1‘), (3, ‘jack1‘), (4, ‘jack1‘)]

  

#取反，查询用户nid不在1,3,4这个列表里的用户信息

ret1=session.query(User.nid,User.username).filter(~User.nid.in_([1,3,4,])).all()

print(ret1)

  

#结果：

[(2, ‘jack2‘), (5, ‘jack2‘)]

 

#查询username=‘jack1‘的所有信息

ret2 = session.query(User.nid,User.username).filter(

User.nid.in_(session.query(User.nid).filter_by(username=‘jack1‘))).all()

print(ret2)

 

#结果：

[(1, ‘jack1‘), (3, ‘jack1‘), (4, ‘jack1‘)]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

#查询nid大于3并且username=‘jack1‘的信息

ret = session.query(User.nid,User.username).filter(and_(User.nid >3,User.username==‘jack1‘)).all()

print(ret)

#结果：

[(4, ‘jack1‘)]

#查询nid小于2或者username等于jack1的数据

ret = session.query(User.nid,User.username).filter(

or_(User.nid < 2, User.username == ‘jack1‘)).all()

print(ret)

#查询用户nid小于2或者username等于jack1并且nid大于3的信息

ret = session.query(User.nid,User.username).filter(

    or_(User.nid < 2,and_(User.username == ‘jack1‘, User.nid > 3))).all()

print(ret)

#结果：

[(1, ‘jack1‘), (4, ‘jack1‘)]

二、通配符


1

2

3

4

5

6

7

8

9

10

11

12

13

#模糊匹配用户名以字母j开头的所有数据

ret = session.query(User.nid,User.username).filter(User.username.like(‘j%‘)).all()

#结果：

[(1, ‘jack1‘), (2, ‘jack2‘), (3, ‘jack1‘), (4, ‘jack1‘), (5, ‘jack2‘)]

#取反

ret1 = session.query(User.nid,User.username).filter(~User.username.like(‘j%‘)).all()

print(ret)

print(ret1)

#结果：

[]

三、限制


1

2

3

4

5

ret=session.query(User.nid,User.username)[1:2]

print(ret)

#结果：

[(2, ‘jack2‘)]

四、排序


1

2

3

4

5

6

7

8

9

10

11

12

13

#倒序排序

ret=session.query(User.nid,User.username).order_by(User.nid.desc()).all()

print(ret)  

#结果：

[(5, ‘jack2‘), (4, ‘jack1‘), (3, ‘jack1‘), (2, ‘jack2‘), (1, ‘jack1‘)]

#正序排序

ret1=session.query(User.nid,User.username).order_by(User.nid.asc()).all()

print(ret1)

#结果：

[(1, ‘jack1‘), (2, ‘jack2‘), (3, ‘jack1‘), (4, ‘jack1‘), (5, ‘jack2‘)]

五、分组


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.sql import func

 

ret = session.query(User.nid,User.username).group_by(User.nid).all()

print(ret)

 

#结果：

[(1, ‘jack1‘), (2, ‘jack2‘), (3, ‘jack1‘), (4, ‘jack1‘), (5, ‘jack2‘)]

 

ret1=session.query(

    func.max(User.nid),

    func.sum(User.nid),

    func.min(User.nid),).group_by(User.username).all()

print(ret1)

 

#结果：

[(4, Decimal(‘8‘), 1), (5, Decimal(‘7‘), 2)]

 

ret2=session.query(

    func.max(User.nid),

    func.sum(User.nid),

    func.min(User.nid), ).group_by(User.username).having(func.min(User.nid)>1).all()

print(ret2)

 

#结果：

[(5, Decimal(‘7‘), 2)]

#打印SQL语句：

from sqlalchemy.sql import func

ret2=session.query(

    func.max(User.nid),

    func.sum(User.nid),

    func.min(User.nid), ).group_by(User.username).having(func.min(User.nid)>1)

print(ret2)

#结果：

SELECT max("user".nid) AS max_1, sum("user".nid) AS sum_1, min("user".nid) AS min_1

FROM "user" GROUP BY "user".username

HAVING min("user".nid) > :min_2

[(‘jack1‘, ‘dba‘), (‘jack2‘, ‘dddd‘)]

SELECT "user".nid AS user_nid, "user".username AS user_username, "user".group_id AS user_group_id

FROM "user" LEFT OUTER JOIN "group" ON "group".nid = "user".group_id

六、组合


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

q1=session.query(User.username).filter(User.nid >2)

q2=session.query(Group.caption).filter(Group.nid <2)

ret = q1.union(q2).all()

print(ret)

#结果：

[(‘jack1‘,), (‘jack2‘,), (‘dba‘,)]

q1=session.query(User.username).filter(User.nid >2)

q2=session.query(Group.caption).filter(Group.nid <2)

ret = q1.union_all(q2).all()

print(ret)

#结果：

[(‘jack1‘,), (‘jack1‘,), (‘jack2‘,), (‘dba‘,)]

二、一对多操作

一对多的关系就需要我们外键来进行约束，下面我们来举例来说明一对多进行连表操作。

1，原始方式：通过join方法来进行连表操作。


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

ret = session.query(User.username,Group.caption).filter(User.nid==Group.nid).all()

print(ret)

  

#结果：

[(‘jack1‘, ‘dba‘), (‘jack2‘, ‘dddd‘)]

  

#通过join来进行连表操作，加isouter的区别：

sql1 = session.query(User).join(Group,isouter=True)

print(sql1)

  

#结果：

SELECT "user".nid AS user_nid, "user".username AS user_username, "user".group_id AS user_group_id

FROM "user" LEFT OUTER JOIN "group" ON "group".nid = "user".group_id

  

sql2 = session.query(User).join(Group)

print(sql2)

  

#结果：

SELECT "user".nid AS user_nid, "user".username AS user_username, "user".group_id AS user_group_id

FROM "user" JOIN "group" ON "group".nid = "user".group_id

#连表操作

ret = session.query(User.username,Group.caption).join(Group,isouter=True).filter(Group.caption == ‘dba‘).all()

print(ret)

#结果：

[(‘jack1‘, ‘dba‘), (‘jack2‘, ‘dba‘), (‘jack1‘, ‘dba‘)]

2，新方式：通过建立relationship的方式


1

2

3

4

5

6

7

#首先在创建表的类中加入relationship字段

class User(Base):

    __tablename__=‘user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    username = Column(String(32))

    group_id = Column(Integer, ForeignKey(‘group.nid‘))

    group = relationship("Group",backref=‘uuu‘)   #跟Group表建立关系，方便查询，backref为虚拟列

正向查询：通过User表查询Group表


1

2

3

4

5

6

7

8

9

10

11

12

13

ret = session.query(User).all()

for obj in ret:

    #obj代指user表的每一行数据

    #obj.group代指group对象

    print(obj.nid,obj.username,obj.group_id,obj.group_id,obj.group,

          obj.group.nid,obj.group.caption)

#结果：

1 jack1 1 1 <__main__.Group object at 0x0000015D762F4630> 1 dba

2 jack2 1 1 <__main__.Group object at 0x0000015D762F4630> 1 dba

3 jack1 2 2 <__main__.Group object at 0x0000015D762F47F0> 2 dddd

4 jack1 1 1 <__main__.Group object at 0x0000015D762F4630> 1 dba

5 jack2 2 2 <__main__.Group object at 0x0000015D762F47F0> 2 dddd

反向查询：通过Group表查询User表


1

2

3

4

5

6

7

8

9

obj = session.query(Group).filter(Group.caption == ‘dba‘).first()

print(obj.nid)

print(obj.caption)

print(obj.uuu)

#结果：

1

dba

[<__main__.User object at 0x000002606096C5C0>, <__main__.User object at 0x000002606096C630>, <__main__.User object at 0x000002606096C6A0>]

我们可以看到上面的例子输出的为对象的列表，输出不太友好，为了达到自己想要的结果，我们可以进行自定义返回结果，请看下面代码，加入__repr__函数：


1

2

3

4

5

6

7

8

9

10

11

12

13

14

class User(Base):

    __tablename__=‘user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    username = Column(String(32))

    group_id = Column(Integer, ForeignKey(‘group.nid‘))

    group = relationship("Group",backref=‘uuu‘)  #跟Group表建立关系，方便查询,常和ForeignKey在一起使用

    def __repr__(self):

        """

        自定义返回结果

        :return:

        """

        temp = ‘%s:%s:%s‘ % (self.nid,self.username,self.group_id)

        return temp

三、多对多操作

1，创建表结构并插入信息


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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

from sqlalchemy import create_engine

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index

from sqlalchemy.orm import sessionmaker, relationship

engine = create_engine("mysql+pymysql://root:[email protected]:3306/mydata", max_overflow=5)

Base = declarative_base()

class Host(Base):

    __tablename__ = ‘host‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    hostname = Column(String(32))

    port = Column(String(32))

    ip = Column(String(32))

class HostUser(Base):

    __tablename__ = ‘host_user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    username = Column(String(32))

#使用for循环时，通过正向反向查询

class HostToHostUser(Base):

    __tablename__ = ‘host_to_host_user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    host_id = Column(Integer,ForeignKey(‘host.nid‘))

    host_user_id = Column(Integer,ForeignKey(‘host_user.nid‘))

    host = relationship("Host",backref=‘h‘)

    host_user = relationship("HostUser",backref=‘u‘)

def init_table():

    """

    创建表，调用Base类的子类

    :return:

    """

    Base.metadata.create_all(engine)

def drop_table():

    Base.metadata.drop_all(engine)

init_table()

Session = sessionmaker(bind=engine)

session = Session()

session.add_all([

    Host(hostname=‘c1‘,port=‘22‘,ip=‘1.1.1.1‘),

    Host(hostname=‘c2‘,port=‘22‘,ip=‘1.1.1.2‘),

    Host(hostname=‘c3‘,port=‘22‘,ip=‘1.1.1.3‘),

    Host(hostname=‘c4‘,port=‘22‘,ip=‘1.1.1.4‘),

    Host(hostname=‘c5‘,port=‘22‘,ip=‘1.1.1.5‘),

])

session.commit()

session.add_all([

    HostUser(username=‘root‘),

    HostUser(username=‘db‘),

    HostUser(username=‘nb‘),

    HostUser(username=‘sb‘),

])

session.commit()

session.add_all([

    HostToHostUser(host_id=1,host_user_id=1),

    HostToHostUser(host_id=1,host_user_id=2),

    HostToHostUser(host_id=1,host_user_id=3),

    HostToHostUser(host_id=2,host_user_id=2),

    HostToHostUser(host_id=2,host_user_id=4),

    HostToHostUser(host_id=2,host_user_id=3),

])

session.commit()

2，需求：获取主机1中所有的用户

 方法一：通过一步一步取


1

2

3

4

5

6

7

8

9

10

11

12

host_obj = session.query(Host).filter(Host.hostname == ‘c1‘).first()

# #取出host_obj.nid

host_to_host_user = session.query(HostToHostUser.host_user_id).filter(HostToHostUser.host_id== host_obj.nid).all()

#

# #因为取出来的结果是[(1,),(2,),(3,)],我们通过内置函数zip来转换成想要的结果

r = zip(*host_to_host_user)

#

users =session.query(HostUser.username).filter(HostUser.nid.in_(list(r)[0])).all()

print(users)

#结果：

[(‘root‘,), (‘db‘,), (‘nb‘,)]

方法二：通过join的方式


1

2

#通过代码整合的代码，相当复杂

session.query(HostUser.name).filter(HostUser.nid.in_(session.query(HostToHostUser.host_user_id).filter(HostToHostUser.host_id == session.query(Host.nid).filter(Host.hostname == ‘c1‘))))

方法三：通过建立relationship的方式

 1，对象


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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

from sqlalchemy import create_engine

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index

from sqlalchemy.orm import sessionmaker, relationship

engine = create_engine("mysql+pymysql://root:[email protected]:3306/mydata", max_overflow=5)

Base = declarative_base()

class HostToHostUser(Base):

    __tablename__ = ‘host_to_host_user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    host_id = Column(Integer,ForeignKey(‘host.nid‘))

    host_user_id = Column(Integer,ForeignKey(‘host_user.nid‘))

class Host(Base):

    __tablename__ = ‘host‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    hostname = Column(String(32))

    port = Column(String(32))

    ip = Column(String(32))

    host_user = relationship(‘HostUser‘,secondary=HostToHostUser.__table__,backref=‘h‘)

# host_user = relationship(‘HostUser‘, secondary=lambda: HostToHostUser.__table__, backref=‘h‘)

#这里加lambda是因为关系表在下面，可以不加lambda,但是关系表要放上面

class HostUser(Base):

    __tablename__ = ‘host_user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    username = Column(String(32))

def init_table():

    """

    创建表，调用Base类的子类

    :return:

    """

    Base.metadata.create_all(engine)

def drop_table():

    Base.metadata.drop_all(engine)

Session = sessionmaker(bind=engine)

session = Session()

host_obj= session.query(Host).filter(Host.hostname==‘c1‘).first()

print(host_obj.host_user)

2，类


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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

from sqlalchemy import create_engine

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index

from sqlalchemy.orm import sessionmaker, relationship

engine = create_engine("mysql+pymysql://root:[email protected]:3306/mydata", max_overflow=5)

Base = declarative_base()

class HostToHostUser(Base):

    __tablename__ = ‘host_to_host_user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    host_id = Column(Integer,ForeignKey(‘host.nid‘))

    host_user_id = Column(Integer,ForeignKey(‘host_user.nid‘))

class Host(Base):

    __tablename__ = ‘host‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    hostname = Column(String(32))

    port = Column(String(32))

    ip = Column(String(32))

    host_user = relationship(‘HostUser‘,secondary=HostToHostUser.__table__,backref=‘h‘)

# host_user = relationship(‘HostUser‘, secondary=lambda: HostToHostUser.__table__, backref=‘h‘)

#这里加lambda是因为关系表在下面，可以不加lambda,但是关系表要放上面

class HostUser(Base):

    __tablename__ = ‘host_user‘

    nid = Column(Integer, primary_key=True,autoincrement=True)

    username = Column(String(32))

def init_table():

    """

    创建表，调用Base类的子类

    :return:

    """

    Base.metadata.create_all(engine)

def drop_table():

    Base.metadata.drop_all(engine)

Session = sessionmaker(bind=engine)

session = Session()

host_obj= session.query(Host).filter(Host.hostname==‘c1‘).first()

print(host_obj.host_user)

今天SQLALchemy就介绍到这里，更多参考信息请参考：

时间： 2024-10-29 19:10:40

SQLAlchmy模块详解的相关文章

Python中time模块详解

在Python中,与时间处理有关的模块就包括:time,datetime以及calendar.这篇文章,主要讲解time模块. 在开始之前,首先要说明这几点: 在Python中,通常有这几种方式来表示时间:1)时间戳 2)格式化的时间字符串 3)元组(struct_time)共九个元素.由于Python的time模块实现主要调用C库,所以各个平台可能有所不同. UTC(Coordinated Universal Time,世界协调时)亦即格林威治天文时间,世界标准时间.在中国为UTC+8.DST

python中threading模块详解（一）

python中threading模块详解(一) 来源 http://blog.chinaunix.net/uid-27571599-id-3484048.html threading提供了一个比thread模块更高层的API来提供线程的并发性.这些线程并发运行并共享内存. 下面来看threading模块的具体用法: 一.Thread的使用 目标函数可以实例化一个Thread对象,每个Thread对象代表着一个线程,可以通过start()方法,开始运行. 这里对使用多线程并发,和不适用多线程并发做

ansible常用模块详解

ansible常用模块详解: ansible <host-pattern> [-m module_name] [-a args] [options] #ansible命令格式  指定主机组或ip地址  指定调用模块   传递给模块的参数   ansible-doc -l #列出可用模块 ansible-doc -s model_name #查看指定模块详细用法 command:ansible默认模块,对指定主机执行命令,不能理解特殊字符 例:ansible web -a 'date' #对we

nginx的配置及模块详解

nginx: nginx是俄罗斯软件工程师Igor Sysoev开发的免费开源web服务器软件,nginx采用了模块化.事件驱动.异步.单线程及非阻塞的架构,并大量采用了多路复用及事件通知机制来实现高并发和高性能,解决C10K的问题,主要功能就是提供http和反向代理服务,以及邮件服务及反向代理等,并且具有多种web服务器功能特性:负载均衡,缓存,访问控制,带宽控制,以及高效整合各种应用的能力. 在nginx中,连接请求由为数不多的几个仅包含一个线程的进程worker以高效的回环(run-loo

python的logging模块详解

日志级别 >>>import logging >>>logging.NOTSET 0 >>>logging.DEBUG 10 >>>logging.INFO 20 >>>logging.WARN 30 >>>logging.ERROR 40 >>>logging.CRITICAL 50 >>>logging._levelNames {0:'NOTSET', 10:

Ceph Monitor基础架构与模块详解

转自:https://www.ustack.com/blog/ceph-monitor/ Ceph rados cluster离不开Monitor,如果没有Monitor,则Ceph将无法执行一条简单的命令.Monitor由于其特殊性,了解它,对于我们深入理解Ceph,保证Ceph的稳定性,有很大帮助. Monitor 基本架构介绍 Monitor的基本架构图: Monitor的主要任务就是维护集群视图的一致性,在维护一致性的时候使用了Paxos协议,并将其实例化到数据库中,方便后续的访问.所以

nodejs QueryString模块 详解

QueryString模块 "QueryString" 模块用于实现URL参数字符串与参数对象的互相转换 此类一共包括4个方法: querystring.stringify(obj, [sep], [eq]) querystring.parse(str, [sep], [eq], [options]) querystring.escape querystring.unescape 一:querystring.stringify(obj,[sep],[eq]) 对象格式化成参数字符串 ,

(转)python time模块和datetime模块详解

python time模块和datetime模块详解 原文:http://www.cnblogs.com/tkqasn/p/6001134.html 一.time模块 time模块中时间表现的格式主要有三种: a.timestamp时间戳,时间戳表示的是从1970年1月1日00:00:00开始按秒计算的偏移量 b.struct_time时间元组,共有九个元素组. c.format time 格式化时间,已格式化的结构使时间更具可读性.包括自定义格式和固定格式. 1.时间格式转换图: 2.主要ti

Ansible安装部署及常用模块详解

Ansible命令使用 Ansible语法使用ansible <pattern_goes_here> -m <module_name> -a <arguments> -m NAME,–module-name=NAME:指定执行使用的模块 -u USERNAME,–user=USERNAME:指定远程主机以USERNAME运行命令 -s,–sudo:相当于Linux系统下的sudo命令 -USUDO_USERNAME,–sudo-user=SUDO_USERNAME:使用