python小白-day11 sqlalchemy

SqlAlchemy ORM　

SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

MySQL-Python

    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>

 

pymysql

    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]

 

MySQL-Connector

    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>

 

cx_Oracle

    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]

 

更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html

步骤一：

使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

#!/usr/bin/env python

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

  

from sqlalchemy import create_engine

  

  

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

  

engine.execute(

    "INSERT INTO ts_test (a, b) VALUES (‘2‘, ‘v1‘)"

)

  

engine.execute(

     "INSERT INTO ts_test (a, b) VALUES (%s, %s)",

    ((555, "v1"),(666, "v1"),)

)

engine.execute(

    "INSERT INTO ts_test (a, b) VALUES (%(id)s, %(name)s)",

    id=999, name="v1"

)

  

result = engine.execute(‘select * from ts_test‘)

result.fetchall()

步骤二：

使用 Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 进行数据库操作。Engine使用Schema Type创建一个特定的结构对象，之后通过SQL Expression Language将该对象转换成SQL语句，然后通过 ConnectionPooling 连接数据库，再然后通过 Dialect 执行SQL，并获取结果。


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

#!/usr/bin/env python

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

metadata = MetaData()

user = Table(‘user‘, metadata,

    Column(‘id‘, Integer, primary_key=True),

    Column(‘name‘, String(20)),

)

color = Table(‘color‘, metadata,

    Column(‘id‘, Integer, primary_key=True),

    Column(‘name‘, String(20)),

)

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

metadata.create_all(engine)

添加（在上述代码后面添加）：


1

2

3

4

5

conn = engine.connect()

# 创建SQL语句，INSERT INTO "user" (id, name) VALUES (:id, :name)

conn.execute(user.insert(),{‘id‘:7,‘name‘:‘seven‘})

conn.close()

删除（同上）：


1

2

3

4

#sql = user.insert().values(id=123, name=‘hetan‘)

#conn.execute(sql)

sql = user.delete().where(user.c.id > 1)

conn.execute(sql)

修改（同上）：


1

2

sql = user.update().where(user.c.name == ‘hetan‘).values(name=‘ed‘)

conn.execute(sql)

查询（同上）：


1

2

3

4

sql = select([user,])

result = conn.execute(sql)

print(result.fetchall())

查询语句还有如下：


1

2

3

4

# sql = select([user.c.id, ])

# sql = select([user.c.name, color.c.name]).where(user.c.id==color.c.id)

# sql = select([user.c.name]).order_by(user.c.name)

# sql = select([user]).group_by(user.c.name)

一个完整的例子：


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

#!/usr/bin/env python

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column ,Integer ,String

from sqlalchemy.orm import sessionmaker

from sqlalchemy import create_engine

Base = declarative_base()

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

class Host(Base):

    __tablename__= ‘hosts‘

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

    hostname = Column(String(64),unique=True,nullable=False)

    ip_addr = Column(String(128),unique=True,nullable=False)

    port = Column(Integer,default=22)

Base.metadata.create_all(engine)

if __name__ == ‘__main__‘:

    SessionCls = sessionmaker(bind=engine)

    session = SessionCls()

    h1 = Host(hostname=‘localhost‘,ip_addr=‘127.0.0.1‘)

    h2 = Host(hostname=‘unbuntu‘,ip_addr=‘192.168.1.1‘)

    session.add_all([h1,h2])

    session.commit()


1

2

3

4

5

h3 = Host(hostname=‘ubuntu2‘,ip_addr=‘192.168.2.244‘,port=20000)

    h3.hostname = ‘ubuntu_test‘  #只要没提交,此时修改也没问题

    session.rollback()

    session.add(h3)

    session.commit()

注：SQLAlchemy无法修改表结构，如果需要可以使用SQLAlchemy开发者开源的另外一个软件Alembic来完成。

步骤三：

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有组件对数据进行操作。根据类创建对象，对象转换成SQL，执行SQL。

创建表：


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

#!/usr/bin/env python

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

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column, Integer, String

from sqlalchemy.orm import sessionmaker

from sqlalchemy import create_engine

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

Base = declarative_base()

class User(Base):

    __tablename__ = ‘users‘

    id = Column(Integer, primary_key=True)

    name = Column(String(50))

# 寻找Base的所有子类，按照子类的结构在数据库中生成对应的数据表信息

Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)

session = Session()

添加：


1

2

3

4

5

6

7

u = User(id=2, name=‘sb‘)

session.add(u)

session.add_all([

    User(id=3, name=‘sb‘),

    User(id=4, name=‘sb‘)

])

session.commit()

删除：


1

2

session.query(User).filter(User.id > 2).delete()

session.commit()

修改:

    

  1. res = session.query(User).filter(User.id == 2)
    2. 

  2. res.id = 3
    3. 

  3. print(res.id)
    4. 

  4. session.commit()
    5.

查询：

    

  1. ret = session.query(User).filter_by(name=‘sb‘).first()
    2. 

  2. print(ret.id)
    3. 

  3. ret = session.query(User).filter_by(name=‘sb‘).all()
    4. 

  4. print(ret)
    5. 

  5. session.commit()
    6.

还有以下查询方法：

    

  1. # ret = session.query(User).filter(User.name.in_([‘sb‘,‘bb‘])).all()
    2. 

  2. # print ret
    3. 

  3.   
    4. 

  4. # ret = session.query(User.name.label(‘name_label‘)).all()
    5. 

  5. # print ret,type(ret)
    6. 

  6.   
    7. 

  7. # ret = session.query(User).order_by(User.id).all()
    8. 

  8. # print ret
    9. 

  9.   
    10. 

  10. # ret = session.query(User).order_by(User.id)[1:3]
    11. 

  11. # print ret
    12.

外键关联

1、单向关联：

    

  1. #!/usr/bin/env python
    2. 

  2. # -*- coding:utf-8 -*-
    3. 

  3. from sqlalchemy.ext.declarative import declarative_base
    4. 

  4. from sqlalchemy import Column ,Integer ,String,ForeignKey,Table
    5. 

  5. from sqlalchemy.orm import sessionmaker,relationship
    6. 

  6. from sqlalchemy import create_engine
    7. 

  7. Base = declarative_base()
    8. 

  8. engine = create_engine("mysql+pymysql://root:[email protected]:3306/test",echo=True)
    9. 

  9. class Host(Base):
    10. 

  10.     __tablename__= ‘hosts‘
    11. 

    12. 

  12.     id = Column(Integer,primary_key=True,autoincrement=True)
    13. 

  13.     group_id = Column(Integer,ForeignKey(‘group.id‘))
    14. 

  14.     hostname = Column(String(64),unique=True,nullable=False)
    15. 

  15.     ip_addr = Column(String(128),unique=True,nullable=False)
    16. 

  16.     port = Column(Integer,default=22)
    17. 

  17.     groups = relationship(‘Group‘,
    18. 

  18.                           backref=‘host‘)
    19. 

  19.     def __repr__(self):
    20. 

  20.         return ‘<id=%s hostname=%s ip_addr=%s>‘ %(self.id,self.hostname,self.ip_addr)
    21. 

  21. class Group(Base):
    22. 

    23. 

  23.     __tablename__ = ‘group‘
    24. 

  24.     id = Column(Integer,primary_key=True)
    25. 

  25.     name = Column(String(64),unique=True,nullable=True)
    26. 

  26.     def __repr__(self):
    27. 

  27.         return ‘<id=%s name=%s>‘ %(self.id,self.name)
    28. 

    29. 

  29. Base.metadata.create_all(engine)
    30. 

  30. SessionCls = sessionmaker(bind=engine)
    31. 

    32. 

  32. session = SessionCls()
    33. 

  33. g1 = Group(name = ‘g1‘)
    34. 

  34. g2 = Group(name = ‘g2‘)
    35. 

  35. g3 = Group(name = ‘g3‘)
    36. 

  36. g4 = Group(name = ‘g4‘)
    37. 

  37. session.add_all([g1,g2,g3,g4])
    38. 

  38. h1 = Host(hostname=‘hetan‘,ip_addr=‘127.0.0.1‘)
    39. 

  39. h2 = Host(hostname=‘liuyao‘,ip_addr=‘10.0.0.1‘)
    40. 

  40. session.add_all([h1,h2])
    41. 

  41. session.commit()
    42.

很明显没关联上，这时可以更新一下，也可以在创建时就关联：

    

  1. g4 = session.query(Group).filter(Group.name==‘g4‘).first()
    2. 

  2. h1 = session.query(Host).filter(Host.hostname==‘hetan‘).update({‘group_id‘:g4.id})
    3. 

  3. session.commit()
    4.

关联成功

2、双向关联：

    

  1. #!/usr/bin/env python
    2. 

  2. from sqlalchemy.ext.declarative import declarative_base
    3. 

  3. from sqlalchemy import Column ,Integer ,String,ForeignKey,Table
    4. 

  4. from sqlalchemy.orm import sessionmaker,relationship
    5. 

  5. from sqlalchemy import create_engine
    6. 

  6. Base = declarative_base()
    7. 

  7. engine = create_engine("mysql+pymysql://root:[email protected]:3306/test",echo=True)
    8. 

  8. host_to_group = Table(
    9. 

  9.     ‘host_2_group‘,Base.metadata,
    10. 

  10.     Column(‘host_id‘,ForeignKey(‘hosts.id‘),primary_key=True),
    11. 

  11.     Column(‘group_id‘,ForeignKey(‘group.id‘),primary_key=True)
    12. 

  12. )
    13. 

  13. class Host(Base):
    14. 

  14.     __tablename__= ‘hosts‘
    15. 

    16. 

  16.     id = Column(Integer,primary_key=True,autoincrement=True)
    17. 

  17.     hostname = Column(String(64),unique=True,nullable=False)
    18. 

  18.     ip_addr = Column(String(128),unique=True,nullable=False)
    19. 

  19.     port = Column(Integer,default=22)
    20. 

  20.     groups = relationship(‘Group‘,
    21. 

  21.                           secondary=host_to_group,
    22. 

  22.                           backref=‘host‘)
    23. 

  23.     def __repr__(self):
    24. 

  24.         return ‘<id=%s hostname=%s ip_addr=%s>‘ %(self.id,self.hostname,self.ip_addr)
    25. 

  25. class Group(Base):
    26. 

    27. 

  27.     __tablename__ = ‘group‘
    28. 

  28.     id = Column(Integer,primary_key=True)
    29. 

  29.     name = Column(String(64),unique=True,nullable=True)
    30. 

  30.     def __repr__(self):
    31. 

  31.         return ‘<id=%s name=%s>‘ %(self.id,self.name)
    32. 

    33. 

  33. Base.metadata.create_all(engine)
    34. 

  34. SessionCls = sessionmaker(bind=engine)
    35. 

    36. 

  36. session = SessionCls()
    37. 

    38. 

  38. g1 = Group(name = ‘g1‘)
    39. 

  39. g2 = Group(name = ‘g2‘)
    40. 

  40. g3 = Group(name = ‘g3‘)
    41. 

  41. g4 = Group(name = ‘g4‘)
    42. 

  42. session.add_all([g1,g2,g3,g4])
    43. 

  43. h1 = Host(hostname=‘hetan‘,ip_addr=‘127.0.0.1‘)
    44. 

  44. h2 = Host(hostname=‘liuyao‘,ip_addr=‘10.0.0.1‘)
    45. 

  45. session.add_all([h1,h2])
    46. 

  46. groups = session.query(Group).all()
    47. 

  47. hosts = session.query(Host).all()
    48. 

  48. print(hosts,groups)
    49. 

  49. h1.groups = groups[1:-1]  #关联
    50. 

  50. g1.host = hosts  #关联
    51. 

  51. session.commit()
    52.

重中之重：

关联结果：

查询：

    

  1. g1 = session.query(Group).first()
    2. 

  2. h1 = session.query(Host).first()
    3. 

  3. print(‘--->‘,g1.host)
    4. 

  4. print(‘--->‘,h1.groups)
    5. 

    6. 

  6. session.commit()
    7.

成功

来自为知笔记(Wiz)

时间： 2024-08-24 10:47:59

python小白-day11 sqlalchemy的相关文章

python小白-day6 time&datetime模块

time&datetime ?一.time模块 time模块提供各种操作时间的函数 说明:一般有两种表示时间的方式:       第一种是时间戳的方式(相对于1970.1.1 00:00:00以秒计算的偏移量),时间戳是惟一的       第二种以数组的形式表示即(struct_time),共有九个元素,分别表示,同一个时间戳的struct_time会因为时区不同而不同 1 2 3 4 5 6 7 8 9 10 11 12 import time print('clock():',time.cl

Python小白-day10 memcache&redis

Memcached Memcached 是一个高性能的分布式内存对象缓存系统,用于动态Web应用以减轻数据库负载.它通过在内存中缓存数据和对象来减少读取数据库的次数,从而提高动态.数据库驱动网站的速度.Memcached基于一个存储键/值对的hashmap.其守护进程(daemon )是用C写的,但是客户端可以用任何语言来编写,并通过memcached协议与守护进程通信. Python操作Memcached 1.第一次操作 import memcache   mc = memcache.Clie

Python小白__网络层级

Python小白__网络分析 刚刚开始接触Python,为了怕遗忘,所以写个博文方便自己回顾,也可以和大家分享,有不同意见,大家共同探讨学习. 网络层级 第一篇是对于网络的一些看法和感想,不只是Python 众做周知,在现在的网络时代,最重要的就是网络通信,通信的方式有甚多 ,有Http,TCP,UDP等等,其不同程序语言实现方式也很多,但是底层的通信原理都相差不多.所以就有了一个网络通信的网络层级之分. 一般网络层级分为四层,分别为:应用层,传输层,网络层,链路层. 假如现在A要传递一条数据给

Python基础day-11[内置函数（未完），递归，匿名函数]

内置函数: abs() : 返回数字的绝对值.参数可以是整数或浮点数,如果参数是复数,则返回复数的模. print(abs(0.2)) print(abs(1)) print(abs(-4)) print(abs(-0.2)) print(abs(3+4j)) 执行结果: D:\Python\Python36-32\python.exe E:/Python/DAY-11/tmp.py 0.2 1 4 0.2 5.0 Process finished with exit code 0 all():

python orm框架SQLAlchemy简单应用（数据库操作）

#_*_coding:utf-8_*_ ''' 创建SQLAlchemy应用基本过程 1.创建连接(和关系数据库创建连接) 2.声明映射文件(python中到类和数据库中的表做一对一的映射,这样就可以通过python中的类对数据库中到表进行操作) 3.创建模式(可以创建表) 4.初始化映射实例 5.创建会话 6.持久化实例对象 ''' from sqlalchemy import create_engine #创建连接 engine = create_engine('mysql://root:1

python操作数据库 - SQLAlchemy

ORM简介 ORM全称是Object Relational Mapping(关系对象映射).实质是将关系数据库中的业务数据用对象的形式表示出来,并通过面向对象的方式将这些对象组织起来,最终在应用中创建一个虚拟对象数据库. 核心就是一个class对应一张表. SQLAlchemy现在已经是python中最流行的ORM框架. ORM优点: 1.隐藏数据库实现,让业务代码只访问对象而不是数据库表 2.良好的数据库操作接口,简单.学习成本低 3.动态数据表映射,表结构改变时,减少代码修改量 4.方便引入

python如何用sqlalchemy操作数据库

工具:mysql  python sqlalchemy ---------------------------------------- 准备工作: 1.安装mysql 如果是window环境请参考 win7安装mysql 如果是linux环境请参考 --稍后补充 2.安装python windows: win7安装python2和python3 linux: linux安装python36 linux python2.6.6以后的版本同样要通过源码包安装,看了下这位博主的这篇文章写的非常详细h

Python 学习笔记 - SQLAlchemy(下)

继续上一篇SQLAlchemy的学习之旅. 多对多表的创建 表Host和表HostUser通过表HostToHostUser关联在一起 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index,Table

Python工具之SQLAlchemy

工具介绍:SQLAlchemy是Python编程语言下的一款ORM框架,该框架建立在数据库API之上,使用关系对象映射进行数据库操作,简言之便是:将对象转换成SQL,然后使用数据API执行SQL并获取执行结果. 工作方式:通过定义类的方式来操作数据库,一个类对应数据库中的一个表,一个类对象对应表中的一行数据,通过实例化类的方式来向数据库表中插入数据,通过对象.字段名来获取表中相应字段的值. 以一个具体的例子来说明: #!/usr/bin/env python # -*- coding:utf-8