1.Python中ORM基础启动连接步骤

目录

1.什么是 ORM？

2. 为什么使用 ORM？

3. 常见 Python ORM 框架

4.orm基础配置六步骤

6. 总结

1.什么是 ORM？

ORM全称是Object-Relational Mapping（对象关系映射）。它是一种编程技术，用于在面向对象的编程语言（如 Python）和关系型数据库（如 MySQL、PostgreSQL、SQLite）之间建立映射关系。它的核心思想是：

• 将数据库中的表（Table）映射为程序中的类（Class）。

• 将表中的行（Row）映射为类的实例（Object）。

• 将表中的列（Column）映射为实例的属性（Attribute）。

2.为什么使用 ORM？

• 简化数据库操作：开发者无需编写复杂的 SQL 语句，直接通过面向对象的方式操作数据。
• 提高开发效率：减少重复的 SQL 代码，专注于业务逻辑。
• 数据库无关性：ORM 框架通常支持多种数据库，切换数据库时无需修改大量代码。
• 避免 SQL 注入：通过参数化查询等机制增强安全性。

3.常见 Python ORM 框架

• SQLAlchemy：功能强大，支持底层 SQL 操作，适合复杂场景。
• Django ORM：Django 框架内置的 ORM，简洁易用。
• Peewee：轻量级 ORM，语法简洁。
• Tortoise ORM：异步 ORM，适合异步应用（如 FastAPI）。

本次以SQLAlchemy和asyncio框架为主

4.orm基础配置步骤

orm基础配置步骤分为6步：

1.安装依赖（library）

2.创建异步数据库引擎

3.创建对象模型（Entity 实体类 领域对象）。每张表都会建立两个字段（create_time, update_time）

4.通过启动fastapi项目，从而建立数据库表。通常情况下使用监听器

5.创建本地会话（Session）工厂对象，再通过这个对象去创建Session从而进行数据库操作

6.第六步：编写依赖系统

第一步：安装依赖（library）

安装 fastapi、sqlalchemy、异步数据库驱动（asyncmy/aiomysql）等库。

pip install fastapi、sqlalchemy、asyncm、aiomysql

第二步：创建异步数据库引擎

使用create_async_engine创建异步引擎，配置数据库连接 URL。

基础代码如下：

engine = create_async_engine( "mysql+asyncmy://账号:密码@localhost:3306/数据库名称?charset=utf8mb4", echo=True #打印执行的 SQL 语句，方便学习调试 #asyncmy：异步 MySQL 驱动（稳定、不报错） )

第三步：创建对象模型（Entity 实体类 领域对象）。每张表都会建立两个字段（create_time, update_time）

定义 ORM 实体模型类映射数据表，规范要求每张表都固定配置 create_time 创建时间和 update_time 更新时间两个公共字段。

基础代码如下：

# 所有【表模型】都必须继承这个 Base，它是 ORM 的核心 Base = declarative_base() # 定义用户表模型（映射数据库中的 user 表） class User(Base): # 对应 MySQL 中的表名：user __tablename__ = "user" # 主键 ID，自增，非空 id = Column(Integer, primary_key=True, autoincrement=True, comment="用户id 主键") # 用户名，不允许为空，唯一约束（不能重复） username = Column(String(50), nullable=False, unique=True, comment="用户名") # 密码，不允许为空 password = Column(String(100), nullable=False, comment="密码") # 创建时间，插入数据时自动填入当前时间 create_time = Column(DateTime, default=func.now(), comment="创建时间") # 更新时间，新增/修改时自动刷新时间 update_time = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间")

第四步：通过启动fastapi项目，从而建立数据库表。通常情况下使用监听器

这是一个 “项目启动监听” —— 只要 FastAPI 一启动，它就自动帮你跑一段代码。

基础代码如下：

# 创建 FastAPI 应用对象（代表整个后端项目） app = FastAPI() # 监听器：当项目【启动时】，自动执行下面的函数 @app.on_event("startup") # 定义异步函数，名字叫 startup（启动） async def startup(): # 创建数据库连接 async with engine.begin() as conn: # 执行：同步创建所有数据库表 await conn.run_sync(Base.metadata.create_all) # 控制台打印提示信息 print("✅ 数据库表创建成功！")

第五步：创建本地会话（Session）工厂对象，再通过这个对象去创建Session从而进行数据库操作

AsyncSessionLocal 是一个异步会话工厂， 用来创建数据库连接会话（AsyncSession）， 绑定了数据库引擎， 并且设置提交后数据不失效， 专门给异步 ORM 操作使用。

基础代码如下：

# 第五步：创建本地会话（Session）工厂对象 AsyncSessionLocal = async_sessionmaker( bind=engine, # 绑定数据库引擎 class_=AsyncSession, # 指定使用异步会话类 expire_on_commit=False # 提交后不失效（非常重要！） )

第六步：编写依赖系统

这是FastAPI 操作数据库的标准依赖项，自动创建会话 → 自动提交 → 出错自动回滚 → 最后自动关闭。让接口代码不用写任何事务代码，超级安全、规范！

基础代码如下：

# 第六步：编写 FastAPI 依赖项（数据库会话依赖系统） async def get_session(): # 通过AsyncSessionLocal创建一个Session对象 async with AsyncSessionLocal() as session: try: yield session # 把会话交给接口使用 await session.commit() # 接口执行完，自动提交事务 except Exception as e: await session.rollback() # 出错了 → 自动回滚（撤销操作） raise # 把错误抛出去 finally: await session.close() # 无论如何 → 自动关闭会话

这6步完成完成之后，一个基础的orm框架就搭好了

6.总结

ORM 是连接 Python 程序与数据库的桥梁，通过面向对象的方式简化数据操作。掌握其核心概念（模型、会话、查询）是使用 Python ORM 框架的基础。不同框架语法略有差异，但核心思想一致。

整和好的完整代码：

from fastapi import FastAPI, Depends from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from sqlalchemy.orm import declarative_base, DeclarativeBase, Mapped from sqlalchemy import Column, Integer, String, DateTime, func, select, update, delete from ferisi2.orm自己小任务.外文件.routers import orm_router engine = create_async_engine( "mysql+asyncmy://root:123456@localhost:3306/zy58?charset=utf8mb4", echo=True ) class Base(DeclarativeBase): c_t =Column(DateTime, default=func.now(), comment='创建时间') u_t = Column(DateTime, default=func.now(), onupdate=func.now(), comment='更新时间') class User(Base): __tablename__="student" id = Column(Integer, primary_key=True,autoincrement=True,comment='用户id 主键') name=Column(String(10),nullable=False,unique=False) age=Column(Integer,nullable=False) gender=Column(String(1),nullable=False) app=FastAPI() @app.on_event("startup") async def startup(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) AsyncSessionLocal=async_sessionmaker(bind=engine,class_=AsyncSession,expire_on_commit=False) async def get_session(): async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception as e: await session.rollback() raise finally: await session.close() if __name__=="__main__": #引入了路由进行增删改查！！！ app.include_router(orm_router) import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)