目录
一、环境
工作中需要用到python和mysql数据库，本次文档记录相关操作。
环境：windows10、python 3.11.7
mysql版本：5.7
二、MySQL的连接和使用
本人使用过的两种方式
2.1方式一：sql为主
2.1.1创建连接
import sqlalchemy
from sqlalchemy.orm import scoped_session, sessionmaker
USERNAME = "root" # 用户名
PASSWORD = "123456" # 密码
ADDR = "localhost" # 连接地址 本地localhost 服务器就是服务器的地址XXX
PORT = "3306" # mysql端口号
DATABASE = "test" # 连接的数据库名
class MysqlSession:
def __init__(self):
self.create_connection()
def create_connection(self):
engine = sqlalchemy.create_engine(
f"mysql+pymysql://{USERNAME}:{PASSWORD}@{ADDR}:{PORT}/{DATABASE}?charset=utf8",
max_overflow=50, # 超过连接池大小外最多创建的连接
pool_size=50, # 连接池大小
pool_recycle=30 # 多久之后对线程池中的线程进行一次连接的回收（重置）,
, echo=False
)
self.connection = scoped_session(sessionmaker(bind=engine))
def get_connection(self):
return self.connection
def ins(self, sql): # 新增数据
return self.connection.execute(sql).lastrowid
def arr(self, sql): # 查询多条数据
return self.connection.execute(sql).fetchall()
def obj(self, sql): # 查询单个数据
return self.connection.execute(sql).fetchone()
def upd(self, sql): # 修改单个数据
return self.connection.execute(sql)
def dlt(self, sql): # 删除数据
return self.connection.execute(sql)
def commit(self):
self.connection.commit() # 提交
self.connection.remove() # 结束会话2.1.2 表结构
2.1.3 新增数据
mysql_session = MysqlSession()
connection = mysql_session.get_connection()
table_name = 'test.user' # 表名（这里是数据库名+表名）
# 新增数据，返回的该条数据的id
name_remark = [
{'name': 'Alice', 'remark': '可能是个女生'},
{'name': 'Bob', 'remark': "可能是个男生"},
{'name': 'Tammi'}
]
new_ids = []
for p in name_remark:
name = p.get('name')
remark = p.get('remark')
sql = f"""insert into {table_name} (name,remark) values ('{name}','{remark}')"""
user_id = mysql_session.ins(sql)
new_ids.append(user_id)
print(new_ids)
mysql_session.commit()
2.1.4 查看数据
# 查看所有数据
sql = f"""select * from {table_name}"""
users = mysql_session.arr(sql)
for u in users:
print(u)
# 查看指定数据
sql = f"""select * from {table_name} where name='Alice'"""
user = mysql_session.obj(sql)
print(user)
2.1.5 修改数据
# 修改指定数据
sql = f"""update {table_name} set name='Alice_changed' where id=16"""
mysql_session.upd(sql)
mysql_session.commit()
# 查看刚才修改的数据
sql = f"""select * from {table_name} where id=16 """
user = mysql_session.obj(sql)
print(user)
2.1.6 删除数据
# 删除指定数据
sql = f"""delete from {table_name} where id=16"""
mysql_session.dlt(sql)
mysql_session.commit()
# 查看所有数据
users = show_test(mysql_session, table_name)
print(users)
2.2方式二：orm对象关系映射
因为现目前工作中没有用到这个（以前用django的时候有用到过orm），这里就简单记录一下测试情况。
2.2.1 mysql连接
mysql_session.py
import sqlalchemy
from sqlalchemy.orm import sessionmaker
USERNAME = "root" # 用户名
PASSWORD = "123456" # 密码
ADDR = "localhost" # 连接地址 本地localhost 服务器就是服务器的地址XXX
PORT = "3306" # mysql端口号
DATABASE = "test" # 连接的数据库名
class MysqlSession:
def __init__(self):
self.create_connection()
def create_connection(self):
self.engine = sqlalchemy.create_engine(
f"mysql+pymysql://{USERNAME}:{PASSWORD}@{ADDR}:{PORT}/{DATABASE}?charset=utf8",
max_overflow=50, # 超过连接池大小外最多创建的连接
pool_size=50, # 连接池大小
pool_recycle=30, # 多久之后对线程池中的线程进行一次连接的回收（重置）,
echo=False
)
Session = sessionmaker(bind=self.engine)
self.session = Session()
def get_session(self):
return self.session
def get_engine(self):
return self.engine
2.2.2 创建表
models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from mysql_session import MysqlSession
Base = declarative_base()
class User(Base):
__tablename__ = 'person'
id = Column(Integer, primary_key=True, autoincrement=True) # 主键 自增
name = Column(String(255))
age = Column(Integer)
engine = MysqlSession().get_engine()
Base.metadata.create_all(engine) #创建上面的表，运行一次即可
2.2.3 新增数据
from models import Person
from mysql_session import MysqlSession
session = MysqlSession().get_session()
# 新增几条数据
new_user = Person(name='张三', age=30)
session.add(new_user)
new_user = Person(name='李四', age=40)
session.add(new_user)
new_user = Person(name='王五', age=50)
session.add(new_user)
session.commit()
2.2.4 查询数据
# 查询所有数据
users = session.query(Person).all()
for user in users:
print(user.name, user.age)
print("**************************************")
# 指定信息
user = session.query(Person).filter_by(name='李四').first()
print(user.name, user.age)
2.2.5 修改数据
# 修改数据
user = session.query(Person).filter_by(name='李四').first()
if user:
user.age = 400 # 将年龄修改为400
session.commit()
print("updated success.")
print(user.name, user.age)
else:
print("not found.")
2.2.6 删除数据
# 删除所有年龄大于100的用户
users = session.query(Person).filter(Person.age > 100).all()
for u in users:
print(u.name, u.age)
for user in users:
session.delete(user)
session.commit()
