当前位置：Gxlcms > Python > python操作SQL

python操作SQL

时间：2021-07-01 10:21:17 帮助过：43人阅读

pymsql是Python中操作MySQL的模块，其使用方法和MySQLdb几乎相同

一、下载安装

pip3 install pymysql

二、操作使用

1、执行SQL

  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  3. import pymysql
  4.   
  5. # 创建连接
  6. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
  7. # 创建游标
  8. cursor = conn.cursor()
  9.   
  10. # 执行SQL，并返回收影响行数
  11. effect_row = cursor.execute("update hosts set host = '1.1.1.2'")
  12.   
  13. # 执行SQL，并返回受影响行数
  14. #effect_row = cursor.execute("update hosts set host = '1.1.1.2' where nid > %s", (1,))
  15.   
  16. # 执行SQL，并返回受影响行数
  17. #effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
  18.   
  19.   
  20. # 提交，不然无法保存新建或者修改的数据
  21. conn.commit()
  22.   
  23. # 关闭游标
  24. cursor.close()
  25. # 关闭连接
  26. conn.close()

2、获取新创建数据自增ID

  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  3. import pymysql
  4.   
  5. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
  6. cursor = conn.cursor()
  7. cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
  8. conn.commit()
  9. cursor.close()
  10. conn.close()
  11.   
  12. # 获取最新自增ID
  13. new_id = cursor.lastrowid

3、获取查询数据

  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  3. import pymysql
  4.   
  5. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
  6. cursor = conn.cursor()
  7. cursor.execute("select * from hosts")
  8.   
  9. # 获取第一行数据
  10. row_1 = cursor.fetchone()
  11.   
  12. # 获取前n行数据
  13. # row_2 = cursor.fetchmany(3)
  14. # 获取所有数据
  15. # row_3 = cursor.fetchall()
  16.   
  17. conn.commit()
  18. cursor.close()
  19. conn.close()

注：在fetch数据时按照顺序进行，可以使用cursor.scroll(num,mode)来移动游标位置，如：

cursor.scroll(1,mode='relative') # 相对当前位置移动

cursor.scroll(2,mode='absolute') # 相对绝对位置移动

4、fetch数据类型关于默认获取的数据是元祖类型，如果想要或者字典类型的数据，即：

  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  3. import pymysql
  4.   
  5. conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
  6.   
  7. # 游标设置为字典类型
  8. cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
  9. r = cursor.execute("call p1()")
  10.   
  11. result = cursor.fetchone()
  12.   
  13. conn.commit()
  14. cursor.close()
  15. conn.close()

SQLAlchemy

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

安装：

pip3 install SQLAlchemy

22.png

SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

  1. MySQL-Python
  2.     mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
  3.    
  4. pymysql
  5.     mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
  6.    
  7. MySQL-Connector
  8.     mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
  9.    
  10. cx_Oracle
  11.     oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]

一、内部处理

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

  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  3. from sqlalchemy import create_engine
  4.   
  5.   
  6. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
  7.   
  8. # 执行SQL
  9. # cur = engine.execute(
  10. #     "INSERT INTO hosts (host, color_id) VALUES ('1.1.1.22', 3)"
  11. # )
  12.   
  13. # 新插入行自增ID
  14. # cur.lastrowid
  15.   
  16. # 执行SQL
  17. # cur = engine.execute(
  18. #     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[('1.1.1.22', 3),('1.1.1.221', 3),]
  19. # )
  20.   
  21.   
  22. # 执行SQL
  23. # cur = engine.execute(
  24. #     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
  25. #     host='1.1.1.99', color_id=3
  26. # )
  27.   
  28. # 执行SQL
  29. # cur = engine.execute('select * from hosts')
  30. # 获取第一行数据
  31. # cur.fetchone()
  32. # 获取第n行数据
  33. # cur.fetchmany(3)
  34. # 获取所有数据
  35. # cur.fetchall()

二、ORM功能使用

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

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, UniqueConstraint, Index
  5. from sqlalchemy.orm import sessionmaker, relationship
  6. from sqlalchemy import create_engine
  7.  
  8. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
  9.  
  10. Base = declarative_base()
  11.  
  12. # 创建单表
  13. class Users(Base):
  14.     __tablename__ = 'users'
  15.     id = Column(Integer, primary_key=True)
  16.     name = Column(String(32))
  17.     extra = Column(String(16))
  18.  
  19.     __table_args__ = (
  20.     UniqueConstraint('id', 'name', name='uix_id_name'),
  21.         Index('ix_id_name', 'name', 'extra'),
  22.     )
  23.  
  24.  
  25. # 一对多
  26. class Favor(Base):
  27.     __tablename__ = 'favor'
  28.     nid = Column(Integer, primary_key=True)
  29.     caption = Column(String(50), default='red', unique=True)
  30.  
  31.  
  32. class Person(Base):
  33.     __tablename__ = 'person'
  34.     nid = Column(Integer, primary_key=True)
  35.     name = Column(String(32), index=True, nullable=True)
  36.     favor_id = Column(Integer, ForeignKey("favor.nid"))
  37.  
  38.  
  39. # 多对多
  40. class Group(Base):
  41.     __tablename__ = 'group'
  42.     id = Column(Integer, primary_key=True)
  43.     name = Column(String(64), unique=True, nullable=False)
  44.     port = Column(Integer, default=22)
  45.  
  46.  
  47. class Server(Base):
  48.     __tablename__ = 'server'
  49.  
  50.     id = Column(Integer, primary_key=True, autoincrement=True)
  51.     hostname = Column(String(64), unique=True, nullable=False)
  52.  
  53.  
  54. class ServerToGroup(Base):
  55.     __tablename__ = 'servertogroup'
  56.     nid = Column(Integer, primary_key=True, autoincrement=True)
  57.     server_id = Column(Integer, ForeignKey('server.id'))
  58.     group_id = Column(Integer, ForeignKey('group.id'))
  59.  
  60.  
  61. def init_db():
  62.     Base.metadata.create_all(engine)
  63.  
  64.  
  65. def drop_db():
  66.     Base.metadata.drop_all(engine)
  67. 注：设置外检的另一种方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])

2、操作表

  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, UniqueConstraint, Index
  5. from sqlalchemy.orm import sessionmaker, relationship
  6. from sqlalchemy import create_engine
  7. engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
  8. Base = declarative_base()
  9. # 创建单表
  10. class Users(Base):
  11.     __tablename__ = 'users'
  12.     id = Column(Integer, primary_key=True)
  13.     name = Column(String(32))
  14.     extra = Column(String(16))
  15.     __table_args__ = (
  16.     UniqueConstraint('id', 'name', name='uix_id_name'),
  17.         Index('ix_id_name', 'name', 'extra'),
  18.     )
  19.     def __repr__(self):
  20.         return "%s-%s" %(self.id, self.name)
  21. # 一对多
  22. class Favor(Base):
  23.     __tablename__ = 'favor'
  24.     nid = Column(Integer, primary_key=True)
  25.     caption = Column(String(50), default='red', unique=True)
  26.     def __repr__(self):
  27.         return "%s-%s" %(self.nid, self.caption)
  28. class Person(Base):
  29.     __tablename__ = 'person'
  30.     nid = Column(Integer, primary_key=True)
  31.     name = Column(String(32), index=True, nullable=True)
  32.     favor_id = Column(Integer, ForeignKey("favor.nid"))
  33.     # 与生成表结构无关，仅用于查询方便
  34.     favor = relationship("Favor", backref='pers')
  35. # 多对多
  36. class ServerToGroup(Base):
  37.     __tablename__ = 'servertogroup'
  38.     nid = Column(Integer, primary_key=True, autoincrement=True)
  39.     server_id = Column(Integer, ForeignKey('server.id'))
  40.     group_id = Column(Integer, ForeignKey('group.id'))
  41.     group = relationship("Group", backref='s2g')
  42.     server = relationship("Server", backref='s2g')
  43. class Group(Base):
  44.     __tablename__ = 'group'
  45.     id = Column(Integer, primary_key=True)
  46.     name = Column(String(64), unique=True, nullable=False)
  47.     port = Column(Integer, default=22)
  48.     # group = relationship('Group',secondary=ServerToGroup,backref='host_list')
  49. class Server(Base):
  50.     __tablename__ = 'server'
  51.     id = Column(Integer, primary_key=True, autoincrement=True)
  52.     hostname = Column(String(64), unique=True, nullable=False)
  53. def init_db():
  54.     Base.metadata.create_all(engine)
  55. def drop_db():
  56.     Base.metadata.drop_all(engine)
  57. Session = sessionmaker(bind=engine)
  58. session = Session()
  1. obj = Users(name="alex0", extra='sb')
  2. session.add(obj)
  3. session.add_all([
  4.     Users(name="alex1", extra='sb'),
  5.     Users(name="alex2", extra='sb'),
  6. ])
  7. session.commit()
  1.  session.query(Users).filter(Users.id > 2).delete()
  2.  session.commit()
  1. session.query(Users).filter(Users.id > 2).update({"name" : "099"})
  2. session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False)
  3. session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate")
  4. session.commit()
  1.  ret = session.query(Users).all()
  2.  ret = session.query(Users.name, Users.extra).all()
  3.  ret = session.query(Users).filter_by(name='alex').all()
  4.  ret = session.query(Users).filter_by(name='alex').first()
  1. ret = session.query(Users).filter_by(name='alex').all()
  2. ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
  3. ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
  4. ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
  5. ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
  6. ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
  7. from sqlalchemy import and_, or_
  8. ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
  9. ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
  10. ret = session.query(Users).filter(
  11.     or_(
  12.         Users.id < 2,
  13.         and_(Users.name == 'eric', Users.id > 3),
  14.         Users.extra != ""
  15.     )).all()
  16. # 通配符
  17. ret = session.query(Users).filter(Users.name.like('e%')).all()
  18. ret = session.query(Users).filter(~Users.name.like('e%')).all()
  19. # 限制
  20. ret = session.query(Users)[1:2]
  21. # 排序
  22. ret = session.query(Users).order_by(Users.name.desc()).all()
  23. ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
  24. # 分组
  25. from sqlalchemy.sql import func
  26. ret = session.query(Users).group_by(Users.extra).all()
  27. ret = session.query(
  28.     func.max(Users.id),
  29.     func.sum(Users.id),
  30.     func.min(Users.id)).group_by(Users.name).all()
  31. ret = session.query(
  32.     func.max(Users.id),
  33.     func.sum(Users.id),
  34.     func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
  35. # 连表
  36. ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
  37. ret = session.query(Person).join(Favor).all()
  38. ret = session.query(Person).join(Favor, isouter=True).all()
  39. # 组合
  40. q1 = session.query(Users.name).filter(Users.id > 2)
  41. q2 = session.query(Favor.caption).filter(Favor.nid < 2)
  42. ret = q1.union(q2).all()
  43. q1 = session.query(Users.name).filter(Users.id > 2)
  44. q2 = session.query(Favor.caption).filter(Favor.nid < 2)
  45. ret = q1.union_all(q2).all()

人气教程排行

本站所有资源全部来源于网络，若本站发布的内容侵害到您的隐私或者利益，请联系我们删除！