AI 助理
备案控制台
开发者社区 数据库 文章 正文

pymysql

2016-08-18 1100
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
RDS MySQL Serverless 基础系列，0.5-2RCU 50GB
推荐场景：
学生管理系统数据库设计 搭建个人博客
RDS MySQL Serverless 高可用系列，价值2615元额度，1个月
云数据库 RDS MySQL，高可用系列 2核4GB
简介： 本篇对于python操作Mysql主要有两种情况 　　·原生模块 pymsql 　　·ORM框架 SQLAchemy pymysql pymsql是python中操作的MYsql的模块,其使用方法和MySQLdb几乎相同 下载安装 　　1丶pip3 install pymsql 使用操作： 　　1·执行SQL   conn= pymsql.

本篇对于python操作Mysql主要有两种情况

　　·原生模块 pymsql

　　·ORM框架 SQLAchemy

pymysql

pymsql是python中操作的MYsql的模块,其使用方法和MySQLdb几乎相同

下载安装

　　1丶pip3 install pymsql

使用操作：

　　1·执行SQL

 

conn= pymsql.connect(host = "127.0.0.1",post = 3306,user = "root",passwd = "123",db = "t1")
#创建连接

cursor = conn.cursor()
#创建游标

effect_row = cursor.execute("updata host set host = "1.1.1.2"")
#执行SQL，并返回受影响行数

effect_row = cursor.execute("update host set = '1.1.1.2' where nid > %s",(1,))
#执行SQL，并返回受影响行数


effect_row = cursor.executemany("insert info host(host,color_id)values(%s,%s)",[("1.1.1.11",1)("1.1.1.11",2)])
#执行SQL，并返回受影响行数

conn.commit()
#提交,不然无法保存新建或者修改的数据


cursor.close()
#关闭游标


conn.close()
#关闭连接

 

　　2丶获取创建数据自增ID

#!/use/bin/evn python
# -*- coding:utf-8 -*-



import pymsql

conn = pymsql.connect(host= "127.0.0.1",port= 3306,user= "root",passwd="123",db="t1")
cursor = conn.cursor()
cursor.executemanty("insert info hosts(host,color_id)values(%s,%s)",[("1.1.1.11",1),("1.1.1.11",2)])
conn.commit()
cursor.colose()
conn.close()



#获取最新自增ID
new_id = cursor.lastrowid

　　3丶获取查询数据

#!/use/bin/env python
# -*- coding:utf-8 -*-

import pymsql

conn = pymsql.connect(host="127.0.0.1",post=3306,user='root',passwd='123',db='t1')
cursor = conn.cursor()
cursor.execute('select * from hosts')

row_1 = cursor.fetchone()
#获取第一行数据

row_2 = cursor.fetchmany(3)
#获取前N行数据

row_3 = cursor.fetchall()
#获取所有数据

conn.commit()

cursor.close()

conn.close()

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

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

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

4丶fetch数据类型

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

#!/use/bin/env python
# -*- coding:utf-8 -*-


import pymsql


conn = pymsql.connect(host='127.0.0.1',port=3306,user='root',passwb='12
3',db='t1')


#游标设置为字典类型
cursor = conn.cursor(cursor=pymsql.cursor.DictCursor)
r = cursor.execute("call p1()")

result = cursor.fetchone()

conn.commit()

cursor.close()

conn.close()

　SQLAchemy　

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

安装：

　　pip3 install SQLAlchemy

 

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

My SQL - Python 
        mysql+mysldb://<user>:<password>@<host>[:<post>]/<dbname>

pymsql
        mysql+pymsql://<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语句。

 

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine
  
  
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
  
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES ('1.1.1.22', 3)"
# )
  
# 新插入行自增ID
# cur.lastrowid
  
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[('1.1.1.22', 3),('1.1.1.221', 3),]
# )
  
  
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
#     host='1.1.1.99', color_id=3
# )
  
# 执行SQL
# cur = engine.execute('select * from hosts')
# 获取第一行数据
# cur.fetchone()
# 获取第n行数据
# cur.fetchmany(3)
# 获取所有数据
# cur.fetchall()
View Code

 

二、ORM功能使用

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

1、创建表

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
 
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
 
Base = declarative_base()
 
# 创建单表
class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))
 
    __table_args__ = (
    UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_id_name', 'name', 'extra'),
    )
 
 
# 一对多
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)
 
 
class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
 
 
# 多对多
class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
 
 
class Server(Base):
    __tablename__ = 'server'
 
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
 
 
class ServerToGroup(Base):
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
 
 
def init_db():
    Base.metadata.create_all(engine)
 
 
def drop_db():
    Base.metadata.drop_all(engine)
View Code

注：设置外检的另一种方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])

2、操作表

 

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)

Base = declarative_base()

# 创建单表
class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

    __table_args__ = (
    UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_id_name', 'name', 'extra'),
    )

    def __repr__(self):
        return "%s-%s" %(self.id, self.name)

# 一对多
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)

    def __repr__(self):
        return "%s-%s" %(self.nid, self.caption)

class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 与生成表结构无关，仅用于查询方便
    favor = relationship("Favor", backref='pers')

# 多对多
class ServerToGroup(Base):
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
    group = relationship("Group", backref='s2g')
    server = relationship("Server", backref='s2g')

class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
    # group = relationship('Group',secondary=ServerToGroup,backref='host_list')


class Server(Base):
    __tablename__ = 'server'

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)




def init_db():
    Base.metadata.create_all(engine)


def drop_db():
    Base.metadata.drop_all(engine)


Session = sessionmaker(bind=engine)
session = Session()
表结构+数据库连接

 

3丶基本操作 

obj = Users(name="alex0", extra='sb')
session.add(obj)
session.add_all([
    Users(name="alex1", extra='sb'),
    Users(name="alex2", extra='sb'),
])
session.commit()
增 
session.query(Users).filter(Users.id > 2).delete()
session.commit()
删 
session.query(Users).filter(Users.id > 2).update({"name" : "099"})
session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False)
session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate")
session.commit()
改 
ret = session.query(Users).all()
ret = session.query(Users.name, Users.extra).all()
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter_by(name='alex').first()
查 
#　条件
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制
ret = session.query(Users)[1:2]

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret = session.query(Person).join(Favor).all()

ret = session.query(Person).join(Favor, isouter=True).all()


# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()
其他

更多功能参见文档，猛击这里下载PDF

文章标签：
云数据库 RDS MySQL 版
Python
SQL
数据库
关系型数据库
MySQL
相关实践学习
每个IT人都想学的“Web应用上云经典架构”实战
本实验从Web应用上云这个最基本的、最普遍的需求出发，帮助IT从业者们通过“阿里云Web应用上云解决方案”，了解一个企业级Web应用上云的常见架构，了解如何构建一个高可用、可扩展的企业级应用架构。
MySQL数据库入门学习
本课程通过最流行的开源数据库MySQL带你了解数据库的世界。 &nbsp; 相关的阿里云产品：云数据库RDS MySQL 版 阿里云关系型数据库RDS（Relational Database Service）是一种稳定可靠、可弹性伸缩的在线数据库服务，提供容灾、备份、恢复、迁移等方面的全套解决方案，彻底解决数据库运维的烦恼。 了解产品详情:&nbsp;https://www.aliyun.com/product/rds/mysql&nbsp;
17603472426
目录
相关文章
站大爷
|
存储 索引 Python
python字典：怎么取出key对应的值
python字典：怎么取出key对应的值
站大爷
692 0
一个物联网卡发烧友
三大运营商那个流量便宜
要确定中国三大运营商（中国移动、中国联通、中国电信）中哪个提供的流量套餐更为便宜，并不是一个可以直接给出固定答案的问题，因为不同的地区、时间、以及用户的具体需求（如通话时长、短信数量、数据流量等）都会影响套餐的价格和性价比。不过，以下是一些通用的方法来比较和选择较为经济的流量套餐：
一个物联网卡发烧友
976 1
2025「AI安全」全球攻防赛小编
|
1天前
|
云安全 人工智能 自然语言处理
AI说的每一句话，都靠谱吗？
阿里云提供AI全栈安全能力，其中针对AI输入与输出环节的安全合规挑战，我们构建了“开箱即用”与“按需增强”相结合的多层次、可配置的内容安全机制。
2025「AI安全」全球攻防赛小编
1619 0
2025「AI安全」全球攻防赛小编
|
5天前
|
存储 人工智能 安全
AI 越智能，数据越危险？
阿里云提供AI全栈安全能力，为客户构建全链路数据保护体系，让企业敢用、能用、放心用
2025「AI安全」全球攻防赛小编
1301 10
阿里云万网官方
|
8天前
|
域名解析 人工智能
【实操攻略】手把手教学，免费领取.CN域名
即日起至2025年12月31日，购买万小智AI建站或云·企业官网，每单可免费领1个.CN域名首年！跟我了解领取攻略吧~
阿里云万网官方
1040 9
阿里云云原生
|
2天前
|
消息中间件 安全 NoSQL
阿里云通过中国信通院首批安全可信中间件评估
近日，由中国信通院主办的 2025（第五届）数字化转型发展大会在京举行。会上，“阿里云应用服务器软件 AliEE”、“消息队列软件 RocketMQ”、“云数据库 Tair”三款产品成功通过中国信通院“安全可信中间件”系列评估，成为首批获此认证的中间件产品。此次评估覆盖安全可信要求、功能完备性、安全防护能力、性能表现、可靠性与可维护性等核心指标，标志着阿里云中间件产品在多架构适配与安全能力上达到行业领先水平。
阿里云云原生
292 192
阿里云云原生
|
2天前
|
安全 Java Android开发
深度解析 Android 崩溃捕获原理及从崩溃到归因的闭环实践
崩溃堆栈全是 a.b.c？Native 错误查不到行号？本文详解 Android 崩溃采集全链路原理，教你如何把“天书”变“说明书”。RUM SDK 已支持一键接入。
阿里云云原生
306 164
HelloYj
|
1天前
|
开发者
「玩透ESA」ESA启用和加速-ER在加速场景中的应用
本文介绍三种配置方法：通过“A鉴权”模板创建函数并设置触发器路由；在ESA上配置回源302跟随；以及自定义响应头。每步均配有详细截图指引，帮助开发者快速完成相关功能设置，提升服务安全性与灵活性。
HelloYj
290 2

热门文章

最新文章

  • 1
    AliOS Things+阿里云 10分钟搭建智慧农业解决方案
  • 2
    从校招生到核心架构师，支付宝研究员李俊奎谈如何成为一名优秀的程序员
  • 3
    Claude官网中文版：在国内使用claude AI的最佳选择！
  • 4
    解决:bash: syntax error near unexpected token `(‘ ...【亲测有效】
  • 5
    上云就上阿里云背后：云计算已成为新旧动能转换关键
  • 6
    58快狗打车基于Hologres的实时离线统一数据服务
  • 7
    Data truncation: Out of range value
  • 8
    【Linux】关于Linux的部分细节与配置文件
  • 9
    Silverlight MMORPG《窝窝世界》游戏视频
  • 10
    Django之admin
  • 1
    基于springboot的智慧家园物业管理系统
    43
  • 2
    Traefik：Go 实现的云原生反向代理，微服务路由自动化利器
    162
  • 3
    货拉拉用户画像基于 Apache Doris 的数据模型设计与实践
    36
  • 4
    程序员的出路：30岁，我们聊聊那些真实的选择
    174
  • 5
    Apache Doris 4.0 版本正式发布：全面升级 AI 与搜索能力，强化离线计算
    58
  • 6
    《游戏测评工具宝典：告别主观评判，用技术逻辑定义专业标准》
    32
  • 7
    从 Snowflake 到 Apache Doris：Planet 实时分析成本直降 80%、查询加速 90 倍
    27
  • 8
    《游戏测评进阶手册：核心工具应用技巧，打造有深度的技术分析》
    39
  • 9
    Doris MCP Server v0.6.0 正式发布
    30
  • 10
    【安全小贴士】创建ECS 后，这 3 个配置千万别漏过
    79

    • 相关电子书

    更多
  • 低代码开发师（初级）实战教程
  • 冬季实战营第三期：MySQL数据库进阶实战
  • 阿里巴巴DevOps 最佳实践手册
    • 下一篇
    让AI单次生成4万字！WriteHERE：开源AI长文写作框架，单次生成超长文本，小说报告一键搞定！