码迷,mamicode.com
首页 > 数据库 > 详细

python小白-day11 sqlalchemy

时间:2016-03-28 23:25:24      阅读:365      评论:0      收藏:0      [点我收藏+]

标签:

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:123456@127.0.0.1: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:123456@localhost: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:123456@127.0.0.1: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:123456@127.0.0.1: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. res.id = 3
  3. print(res.id)
  4. session.commit()
技术分享
查询:
  1. ret = session.query(User).filter_by(name=‘sb‘).first()
  2. print(ret.id)
  3. ret = session.query(User).filter_by(name=‘sb‘).all()
  4. print(ret)
  5. session.commit()
技术分享
还有以下查询方法:
  1. # ret = session.query(User).filter(User.name.in_([‘sb‘,‘bb‘])).all()
  2. # print ret
  3. # ret = session.query(User.name.label(‘name_label‘)).all()
  4. # print ret,type(ret)
  5. # ret = session.query(User).order_by(User.id).all()
  6. # print ret
  7. # ret = session.query(User).order_by(User.id)[1:3]
  8. # print ret

外键关联

1、单向关联:
  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  3. from sqlalchemy.ext.declarative import declarative_base
  4. from sqlalchemy import Column ,Integer ,String,ForeignKey,Table
  5. from sqlalchemy.orm import sessionmaker,relationship
  6. from sqlalchemy import create_engine
  7. Base = declarative_base()
  8. engine = create_engine("mysql+pymysql://root:123456@localhost:3306/test",echo=True)
  9. class Host(Base):
  10. __tablename__= ‘hosts‘
  11. id = Column(Integer,primary_key=True,autoincrement=True)
  12. group_id = Column(Integer,ForeignKey(‘group.id‘))
  13. hostname = Column(String(64),unique=True,nullable=False)
  14. ip_addr = Column(String(128),unique=True,nullable=False)
  15. port = Column(Integer,default=22)
  16. groups = relationship(‘Group‘,
  17. backref=‘host‘)
  18. def __repr__(self):
  19. return ‘<id=%s hostname=%s ip_addr=%s>‘ %(self.id,self.hostname,self.ip_addr)
  20. class Group(Base):
  21. __tablename__ = ‘group‘
  22. id = Column(Integer,primary_key=True)
  23. name = Column(String(64),unique=True,nullable=True)
  24. def __repr__(self):
  25. return ‘<id=%s name=%s>‘ %(self.id,self.name)
  26. Base.metadata.create_all(engine)
  27. SessionCls = sessionmaker(bind=engine)
  28. session = SessionCls()
  29. g1 = Group(name = ‘g1‘)
  30. g2 = Group(name = ‘g2‘)
  31. g3 = Group(name = ‘g3‘)
  32. g4 = Group(name = ‘g4‘)
  33. session.add_all([g1,g2,g3,g4])
  34. h1 = Host(hostname=‘hetan‘,ip_addr=‘127.0.0.1‘)
  35. h2 = Host(hostname=‘liuyao‘,ip_addr=‘10.0.0.1‘)
  36. session.add_all([h1,h2])
  37. session.commit()
技术分享
很明显没关联上,这时可以更新一下,也可以在创建时就关联:
  1. g4 = session.query(Group).filter(Group.name==‘g4‘).first()
  2. h1 = session.query(Host).filter(Host.hostname==‘hetan‘).update({‘group_id‘:g4.id})
  3. session.commit()
技术分享
关联成功
2、双向关联:
  1. #!/usr/bin/env python
  2. from sqlalchemy.ext.declarative import declarative_base
  3. from sqlalchemy import Column ,Integer ,String,ForeignKey,Table
  4. from sqlalchemy.orm import sessionmaker,relationship
  5. from sqlalchemy import create_engine
  6. Base = declarative_base()
  7. engine = create_engine("mysql+pymysql://root:123456@localhost:3306/test",echo=True)
  8. host_to_group = Table(
  9. ‘host_2_group‘,Base.metadata,
  10. Column(‘host_id‘,ForeignKey(‘hosts.id‘),primary_key=True),
  11. Column(‘group_id‘,ForeignKey(‘group.id‘),primary_key=True)
  12. )
  13. class Host(Base):
  14. __tablename__= ‘hosts‘
  15. id = Column(Integer,primary_key=True,autoincrement=True)
  16. hostname = Column(String(64),unique=True,nullable=False)
  17. ip_addr = Column(String(128),unique=True,nullable=False)
  18. port = Column(Integer,default=22)
  19. groups = relationship(‘Group‘,
  20. secondary=host_to_group,
  21. backref=‘host‘)
  22. def __repr__(self):
  23. return ‘<id=%s hostname=%s ip_addr=%s>‘ %(self.id,self.hostname,self.ip_addr)
  24. class Group(Base):
  25. __tablename__ = ‘group‘
  26. id = Column(Integer,primary_key=True)
  27. name = Column(String(64),unique=True,nullable=True)
  28. def __repr__(self):
  29. return ‘<id=%s name=%s>‘ %(self.id,self.name)
  30. Base.metadata.create_all(engine)
  31. SessionCls = sessionmaker(bind=engine)
  32. session = SessionCls()
  33. g1 = Group(name = ‘g1‘)
  34. g2 = Group(name = ‘g2‘)
  35. g3 = Group(name = ‘g3‘)
  36. g4 = Group(name = ‘g4‘)
  37. session.add_all([g1,g2,g3,g4])
  38. h1 = Host(hostname=‘hetan‘,ip_addr=‘127.0.0.1‘)
  39. h2 = Host(hostname=‘liuyao‘,ip_addr=‘10.0.0.1‘)
  40. session.add_all([h1,h2])
  41. groups = session.query(Group).all()
  42. hosts = session.query(Host).all()
  43. print(hosts,groups)
  44. h1.groups = groups[1:-1] #关联
  45. g1.host = hosts #关联
  46. session.commit()
重中之重:
技术分享
关联结果:
技术分享
查询:
  1. g1 = session.query(Group).first()
  2. h1 = session.query(Host).first()
  3. print(‘--->‘,g1.host)
  4. print(‘--->‘,h1.groups)
  5. session.commit()
技术分享

成功





python小白-day11 sqlalchemy

标签:

原文地址:http://www.cnblogs.com/hetan/p/5331127.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!