当前位置：首页 > article >正文

python sqlalchemy 操作数据库

article 2024/11/25 1:02:02

本篇文章分享使用python 的sqlalchemy框架操作数据库的方法，总共分三个步骤，第一步链接数据库，第二步定义entity，第三步操作数据库，下面是完整的演示

第一步，链接数据库，它在名叫MysqlUtils的python文件中

# 数据库操作
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# 本地
username = 'root'
password = 'root'
host = 'localhost'
port = '3306'
database = 'test'

# 创建连接引擎
engine = create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}',
                       poolclass=QueuePool,
                       pool_size=10,
                       pool_timeout=30)

第二步定义entity，我这个entity叫persion，除了id外，还有两个属性，username和passwd，它在名叫models的python文件中

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Person(Base):
    def __init__(self, username, passwd):
        self.username = username
        self.passwd = passwd

    __tablename__ = 'person'
    id = Column(Integer, primary_key=True)
    username = Column(String)
    passwd = Column(String)

第三步，操作数据库

from sqlalchemy.orm import sessionmaker
from sqlalchemy import func
from model import models
from utils.MysqlUtils import *
from sqlalchemy.sql.expression import text

Session = sessionmaker(bind=engine)
session = Session()


# 插入数据
def insert():
    p = models.Person("ti", "ti")
    session.add(p)
    session.commit()
    session.close()


# 查询总数
def query_total_count():
    total_count = session.query(func.count(models.Person)).scalar()
    # 关闭会话
    session.close()
    return total_count


# 查询所有
def query_all_data():
    # 查询所有用户
    persons = session.query(models.Person).all()
    for person in persons:
        print(f"Person ID: {person.id}, Name: {person.username}, Passwd: {person.passwd}")
    # 关闭会话
    session.close()


# 条件查询
def paginate_query(page_number, page_size):
    # 计算跳过的记录数
    offset = (page_number - 1) * page_size
    # 执行分页查询，这里假设User模型有name字段且非空
    persons = session.query(models.Person).filter(models.Person.username != 'abc').order_by(models.Person.id).offset(
        offset).limit(page_size).all()
    for person in persons:
        print(f"Person ID: {person.id}, Name: {person.username}, Passwd: {person.passwd}")


# 执行一个sql
def query_condition(sql):
    # 执行sql并获取所有结果
    result = session.execute(text(sql)).fetchall()
    print(result)
    persons = [models.Person(row[1], row[2]) for row in result]
    print(persons)
    # 关闭会话
    session.close()


# paginate_query(1, 10)
# insert()
query_all_data()
# query_condition('SELECT * FROM Person')