MySQL : pymysql et SQLAlchemy

Table des matières

1. pymysql

2. SQLAlchemy

1. pymysql

pymsql est un module pour faire fonctionner MySQL en Python. l'utilisation est la suivante MySQLdb est presque la même.

1. Téléchargez et installez

#在终端直接运行
pip3 install pymysql

2. Utilisez l'opération

a. 🎜 >

b. Obtenez l'ID d'auto-incrémentation des données nouvellement créées

#!/usr/bin/env python# -*- coding:utf-8 -*-import pymysql
  # 创建连接conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')# 创建游标cursor = conn.cursor()
  # 执行SQL，并返回受影响行数effect_row = cursor.execute("update hosts set host = '1.1.1.2'")
  # 执行SQL，并返回受影响行数#effect_row = cursor.execute("update hosts set host = '1.1.1.2' where nid > %s", (1,))  # 执行SQL，并返回受影响行数#effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])  
  # 提交，不然无法保存新建或者修改的数据conn.commit()
  # 关闭游标cursor.close()# 关闭连接conn.close()

c Obtenez les données de requête

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

cursor.close()
conn.close()

Remarque : lors de la récupération des données, procédez dans l'ordre. Vous pouvez utiliser curseur.scroll(num, mode) pour déplacer la position du curseur, par exemple :

#!/usr/bin/env python# -*- coding:utf-8 -*-import pymysql
  
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor = conn.cursor()
cursor.execute("select * from hosts")
  # 获取第一行数据row_1 = cursor.fetchone()
  # 获取前n行数据# row_2 = cursor.fetchmany(3)# 获取所有数据# row_3 = cursor.fetchall()  
conn.commit()
cursor.close()
conn.close()

cursor.scroll(1, mode='relative') # Se déplacer par rapport à la position actuelle

• cursor.scroll(2,mode='absolute') 　# Se déplacer par rapport à la position absolue

• d. récupérer le type de données

Les données par défaut obtenues sont de type tuple Si vous souhaitez obtenir des données de type dictionnaire, c'est-à-dire :

2. SQLAlchemy

SQLAlchemy est un framework ORM sous le langage de programmation Python Le framework est construit sur l'API de base de données et utilise le mappage d'objets relationnels pour effectuer des opérations de base de données. , puis utilisez l'API de données pour exécuter SQL et obtenir les résultats d'exécution.

#!/usr/bin/env python# -*- coding:utf-8 -*-import pymysql
  
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
  # 游标设置为字典类型cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
r = cursor.execute("call p1()")
  
result = cursor.fetchone()
  
conn.commit()
cursor.close()
conn.close()

1. Téléchargez et installez

2. Dépendances de SQLAlchemy

SQLAlchemy lui-même ne peut pas faire fonctionner la base de données, il doit s'appuyer sur des tiers. des plug-ins tels que pymsql , Dialect est utilisé pour communiquer avec l'API de données et appeler différentes API de base de données en fonction de différents fichiers de configuration pour réaliser des opérations sur la base de données.

#在终端直接运行pip3 install SQLAlchemy

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...]

3. La fonction ORM utilise 更多详见：indexUtiliser tous les composants ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect pour opérer sur les données. Créez des objets basés sur des classes, convertissez des objets en SQL et exécutez SQL. .html

a. Créer une table

Remarque : Une autre façon de définir des clés étrangères est ForeignKeyConstraint(['other_id'], [ ' othertable.other_id'])

#!/usr/bin/env python# -*- coding:utf-8 -*-from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_engine#表明依赖关系并创建连接，最大连接数为5 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,autoincrement=True)    # id列，主键自增    name = Column(String(32))    # name列    extra = Column(String(16))    # extra列 
    __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)

b. Table d'opération

Table structure + connexion à la base de données

#!/usr/bin/env python# -*- coding:utf-8 -*-from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom 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()

b.1 Ajouter

b.2 Supprimer

#单条增加obj = Users(name="alex0", extra='sb')
session.add(obj)#多条增加session.add_all([
    Users(name="alex1", extra='sb'),
    Users(name="alex2", extra='sb'),
])#提交session.commit()

b.3 Changer

#先查询到要删除的记录，再deletesession.query(Users).filter(Users.id > 2).delete()
session.commit()

b.4 Vérifier

#先查询，再更新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()

b.5 Autres

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(text("id<:value and name=:name")).params(value=224, name=&#39;fred&#39;).order_by(User.id).all()

ret = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name=&#39;ed&#39;).all()

#　条件ret = session.query(Users).filter_by(name='alex').all()    # 条件内为关键字表达式ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()    # 条件内为SQL表达式ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()    # betweenret = session.query(Users).filter(Users.id.in_([1,3,4])).all()    # inret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()    # not inret = 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()    # andret = session.query(Users).filter(or_(Users.id < 2, Users.name == &#39;eric&#39;)).all()    # orret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == &#39;eric&#39;, Users.id > 3),
        Users.extra != "")).all()# 通配符ret = session.query(Users).filter(Users.name.like('e%')).all()    # e开头ret = session.query(Users).filter(~Users.name.like('e%')).all()    # 非e开头# 限制ret = session.query(Users)[1:2]    # 相当于limit# 排序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()    # 默认内连 inner joinret = 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()

Documents de référence :

1. Développement Python [Partie 19] : Opération Python MySQL

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!