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
Copier après la connexion
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()Copier après la connexion
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()Copier après la connexion
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()Copier après la connexion
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
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()Copier après la connexion
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
Copier après la connexion
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...]
Copier après la connexion
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)Copier après la connexion
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()Copier après la connexion
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()
Copier après la connexion
b.3 Changer
#先查询到要删除的记录，再deletesession.query(Users).filter(Users.id > 2).delete()
session.commit()
Copier après la connexion
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()Copier après la connexion
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='fred').order_by(User.id).all()
ret = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()Copier après la connexion
# 条件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 == 'eric')).all() # orret = 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() # 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()Copier après la connexion
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!
![[Web front-end] Démarrage rapide de Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
