权威的南通网站建设,深圳高端家政服务公司,网站备案 影响,西安seo排名优化推广价格SQLModel 系统性指南
目录
简介 什么是 SQLModel#xff1f;为什么使用 SQLModel#xff1f; 安装快速入门 定义模型创建数据库和表 基本 CRUD 操作 创建#xff08;Create#xff09;读取#xff08;Read#xff09;更新#xff08;Update#xff09;删除#xff0…SQLModel 系统性指南
目录
简介 什么是 SQLModel为什么使用 SQLModel 安装快速入门 定义模型创建数据库和表 基本 CRUD 操作 创建Create读取Read更新Update删除Delete 处理关系 一对多关系多对多关系 高级功能 异步支持自定义查询迁移Migrations 与 FastAPI 的集成 依赖注入路由保护 性能优化与最佳实践常见问题解答参考资料 1. 简介
什么是 SQLModel
SQLModel 是一个现代化的 Python 库旨在简化与数据库的交互。它结合了 Pydantic 和 SQLAlchemy 的优势使得定义数据模型、进行数据验证和与数据库交互变得更加直观和高效。SQLModel 由 Sebastián RamírezFastAPI 的创始人开发专为与 FastAPI 框架无缝集成而设计。
为什么使用 SQLModel
简洁性通过结合 Pydantic 的数据验证和 SQLAlchemy 的 ORM 功能SQLModel 使模型定义和数据库操作更加简洁。类型安全充分利用 Python 的类型提示增强代码的可读性和可靠性。与 FastAPI 无缝集成优化了与 FastAPI 的集成支持自动文档生成和依赖注入。灵活性支持同步和异步操作适应不同的性能需求。现代化设计采用现代化的 Python 编码风格和最佳实践提升开发体验。 2. 安装
首先确保您已经安装了 Python 3.7 或更高版本。然后使用 pip 安装 sqlmodel 包
pip install sqlmodel此外根据您使用的数据库还需要安装相应的数据库驱动。例如 SQLite无需额外安装驱动Python 内置支持。 PostgreSQL pip install asyncpgMySQL pip install pymysql3. 快速入门
定义模型
使用 SQLModel 定义数据模型时通常会继承自 SQLModel 并使用 tableTrue 参数指示这是一个数据库表。
from typing import Optional
from sqlmodel import SQLModel, Field
from datetime import datetimeclass User(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)username: str Field(indexTrue, nullableFalse, uniqueTrue)email: str Field(indexTrue, nullableFalse, uniqueTrue)hashed_password: str Field(nullableFalse)is_active: bool Field(defaultTrue)created_at: datetime Field(default_factorydatetime.utcnow)创建数据库和表
使用 SQLAlchemy 的引擎和 SQLModel 的元数据来创建数据库和表。
from sqlmodel import SQLModel, create_engine
from models import User # 假设上面的模型保存在 models.py 文件中DATABASE_URL sqlite:///./test.db # 或者使用其他数据库如 PostgreSQL
engine create_engine(DATABASE_URL, echoTrue)def create_db_and_tables():SQLModel.metadata.create_all(engine)在应用启动时调用 create_db_and_tables 来创建数据库表。 4. 基本 CRUD 操作
创建Create
向数据库中插入一条新记录。
from sqlmodel import Session, select
from models import User
from database import engine, create_db_and_tablesdef create_user(username: str, email: str, hashed_password: str) - User:user User(usernameusername, emailemail, hashed_passwordhashed_password)with Session(engine) as session:session.add(user)session.commit()session.refresh(user)return user读取Read
从数据库中查询记录。
def get_user_by_id(user_id: int) - Optional[User]:with Session(engine) as session:user session.get(User, user_id)return userdef get_user_by_username(username: str) - Optional[User]:with Session(engine) as session:statement select(User).where(User.username username)user session.exec(statement).first()return user更新Update
更新数据库中的记录。
def update_user_email(user_id: int, new_email: str) - Optional[User]:with Session(engine) as session:user session.get(User, user_id)if user:user.email new_emailsession.add(user)session.commit()session.refresh(user)return userreturn None删除Delete
从数据库中删除记录。
def delete_user(user_id: int) - bool:with Session(engine) as session:user session.get(User, user_id)if user:session.delete(user)session.commit()return Truereturn False5. 处理关系
一对多关系
例如一个用户可以有多条地址记录。
from typing import List, Optional
from sqlmodel import SQLModel, Field, Relationshipclass Address(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)street: strcity: struser_id: int Field(foreign_keyuser.id)user: Optional[User] Relationship(back_populatesaddresses)class User(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)username: str Field(indexTrue, nullableFalse, uniqueTrue)email: str Field(indexTrue, nullableFalse, uniqueTrue)hashed_password: str Field(nullableFalse)is_active: bool Field(defaultTrue)created_at: datetime Field(default_factorydatetime.utcnow)addresses: List[Address] Relationship(back_populatesuser)多对多关系
例如用户和角色之间的多对多关系。
from typing import List, Optional
from sqlmodel import SQLModel, Field, Relationshipclass UserRoleLink(SQLModel, tableTrue):user_id: int Field(foreign_keyuser.id, primary_keyTrue)role_id: int Field(foreign_keyrole.id, primary_keyTrue)class Role(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)name: strusers: List[User] Relationship(back_populatesroles,link_modelUserRoleLink)class User(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)username: str Field(indexTrue, nullableFalse, uniqueTrue)email: str Field(indexTrue, nullableFalse, uniqueTrue)hashed_password: str Field(nullableFalse)is_active: bool Field(defaultTrue)created_at: datetime Field(default_factorydatetime.utcnow)roles: List[Role] Relationship(back_populatesusers,link_modelUserRoleLink)6. 高级功能
异步支持
SQLModel 支持异步数据库操作适用于需要高并发和高性能的应用。
首先安装异步驱动如 asyncpg 用于 PostgreSQL
pip install asyncpg然后配置异步引擎和会话
from sqlmodel import SQLModel, create_engine, select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from models import User
from datetime import datetimeDATABASE_URL postgresqlasyncpg://user:passwordlocalhost/dbname
async_engine create_async_engine(DATABASE_URL, echoTrue)async_session sessionmaker(async_engine, class_AsyncSession, expire_on_commitFalse
)async def init_db():async with async_engine.begin() as conn:await conn.run_sync(SQLModel.metadata.create_all)# 在应用启动时调用 init_db
import asyncio
asyncio.run(init_db())# 异步获取会话
async def get_async_session():async with async_session() as session:yield session# 异步 CRUD 操作示例
async def get_user_async(user_id: int) - Optional[User]:async with async_session() as session:user await session.get(User, user_id)return user自定义查询
使用 SQLAlchemy 的强大查询功能执行复杂的数据库操作。
from sqlmodel import Session, select, func
from models import Userdef count_users() - int:with Session(engine) as session:statement select(func.count(User.id))count session.exec(statement).one()return countdef get_users_with_email_domain(domain: str) - List[User]:with Session(engine) as session:statement select(User).where(User.email.like(f%{domain}))users session.exec(statement).all()return users迁移Migrations
虽然 SQLModel 本身不提供迁移工具但它与 Alembic 完全兼容可以使用 Alembic 进行数据库迁移。
安装 Alembic
pip install alembic初始化 Alembic
alembic init alembic配置 Alembic
编辑 alembic.ini设置 sqlalchemy.url 为您的数据库 URL。
在 alembic/env.py 中导入您的模型
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from sqlmodel import SQLModel
import sys
import os# 将项目路径添加到 sys.path
sys.path.append(os.path.dirname(os.path.dirname(__file__)))from models import User # 导入您的模型# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config context.config# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)target_metadata SQLModel.metadatadef run_migrations_offline():...# 保持默认配置def run_migrations_online():...# 保持默认配置if context.is_offline_mode():run_migrations_offline()
else:run_migrations_online()创建迁移脚本
alembic revision --autogenerate -m Initial migration应用迁移
alembic upgrade head7. 与 FastAPI 的集成
依赖注入
利用 FastAPI 的依赖注入机制将数据库会话注入到路由中。
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from models import User
from database import engine, get_sessionapp FastAPI()app.post(/users/, response_modelUser)
def create_user(user: User, session: Session Depends(get_session)):db_user session.exec(select(User).where(User.username user.username)).first()if db_user:raise HTTPException(status_code400, detailUsername already exists)session.add(user)session.commit()session.refresh(user)return userapp.get(/users/{user_id}, response_modelUser)
def read_user(user_id: int, session: Session Depends(get_session)):user session.get(User, user_id)if not user:raise HTTPException(status_code404, detailUser not found)return user路由保护
结合 JWT 进行身份验证保护特定路由。
安装 fastapi-jwt-auth
pip install fastapi-jwt-auth配置 JWT
from fastapi import FastAPI, Depends, HTTPException
from fastapi_jwt_auth import AuthJWT
from pydantic import BaseModel
from sqlmodel import Session, select
from models import User
from database import engine, get_sessionclass Settings(BaseModel):authjwt_secret_key: str your-secret-keyapp FastAPI()AuthJWT.load_config
def get_config():return Settings()app.post(/login)
def login(user: User, Authorize: AuthJWT Depends()):# 验证用户凭证此处省略具体验证逻辑access_token Authorize.create_access_token(subjectuser.username)return {access_token: access_token}app.get(/protected)
def protected(Authorize: AuthJWT Depends()):Authorize.jwt_required()current_user Authorize.get_jwt_subject()return {message: fHello, {current_user}}8. 性能优化与最佳实践
8.1 使用连接池
优化数据库连接使用连接池以提高性能和资源利用率。
from sqlmodel import create_engineDATABASE_URL sqlite:///./test.db
engine create_engine(DATABASE_URL, echoTrue, pool_size20, max_overflow0)8.2 异步操作
对于高并发应用使用异步数据库操作。
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmakerDATABASE_URL postgresqlasyncpg://user:passwordlocalhost/dbname
async_engine create_async_engine(DATABASE_URL, echoTrue)
async_session sessionmaker(async_engine, class_AsyncSession, expire_on_commitFalse
)8.3 缓存
使用缓存机制如 Redis减少数据库查询提高响应速度。
import redisredis_client redis.Redis(hostlocalhost, port6379, db0)def get_user_cached(user_id: int) - Optional[User]:cached_user redis_client.get(fuser:{user_id})if cached_user:return User.parse_raw(cached_user)with Session(engine) as session:user session.get(User, user_id)if user:redis_client.set(fuser:{user_id}, user.json(), ex3600)return user8.4 索引优化
为常用查询字段添加索引提高查询性能。
class User(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)username: str Field(indexTrue, nullableFalse, uniqueTrue)email: str Field(indexTrue, nullableFalse, uniqueTrue)# 其他字段...8.5 分页
对于大量数据查询使用分页机制减少单次查询的数据量。
def get_users_paginated(skip: int 0, limit: int 10) - List[User]:with Session(engine) as session:statement select(User).offset(skip).limit(limit)users session.exec(statement).all()return users9. 常见问题解答
9.1 如何在 SQLModel 中使用外键
在定义模型时使用 Field 的 foreign_key 参数指定外键。
class Address(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)street: strcity: struser_id: int Field(foreign_keyuser.id)user: Optional[User] Relationship(back_populatesaddresses)9.2 SQLModel 支持哪些数据库
SQLModel 基于 SQLAlchemy支持所有 SQLAlchemy 支持的数据库包括
SQLitePostgreSQLMySQLSQL ServerOracle以及其他数据库通过相应的数据库驱动支持。
9.3 如何进行数据库迁移
SQLModel 本身不提供迁移工具但可以与 Alembic 配合使用进行数据库迁移。
安装 Alembic
pip install alembic初始化 Alembic
alembic init alembic配置 Alembic
编辑 alembic.ini设置 sqlalchemy.url 为您的数据库 URL。
在 alembic/env.py 中导入您的模型
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from sqlmodel import SQLModel
import sys
import os# 将项目路径添加到 sys.path
sys.path.append(os.path.dirname(os.path.dirname(__file__)))from models import User # 导入您的模型config context.configfileConfig(config.config_file_name)target_metadata SQLModel.metadatadef run_migrations_offline():...def run_migrations_online():...if context.is_offline_mode():run_migrations_offline()
else:run_migrations_online()创建迁移脚本
alembic revision --autogenerate -m Initial migration应用迁移
alembic upgrade head9.4 如何处理模型验证错误
SQLModel 结合了 Pydantic 的数据验证功能可以在模型定义中使用 Pydantic 的字段验证器。
from sqlmodel import SQLModel, Field
from pydantic import validator, EmailStrclass User(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)username: str Field(indexTrue, nullableFalse, uniqueTrue)email: EmailStr Field(indexTrue, nullableFalse, uniqueTrue)hashed_password: str Field(nullableFalse)validator(username)def username_must_not_be_empty(cls, v):if not v or not v.strip():raise ValueError(Username must not be empty)return v10. 参考资料
SQLModel 官方文档https://sqlmodel.tiangolo.com/SQLAlchemy 官方文档https://www.sqlalchemy.org/FastAPI 官方文档https://fastapi.tiangolo.com/Alembic 官方文档https://alembic.sqlalchemy.org/en/latest/Real Python 的 SQLModel 教程https://realpython.com/sqlmodel-python-orm/Pydantic 官方文档https://pydantic-docs.helpmanual.io/GitHub 上的 SQLModel 仓库https://github.com/tiangolo/sqlmodel 附录完整示例
以下是一个完整的 FastAPI 应用示例展示了如何使用 SQLModel 进行数据库操作和 API 构建。
目录结构
my_fastapi_app/
├── main.py
├── models.py
├── database.py
├── schemas.py
└── alembic/├── env.py├── script.py.mako└── versions/models.py
from typing import List, Optional
from sqlmodel import SQLModel, Field, Relationship
from datetime import datetimeclass Address(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)street: strcity: struser_id: int Field(foreign_keyuser.id)user: Optional[User] Relationship(back_populatesaddresses)class User(SQLModel, tableTrue):id: Optional[int] Field(defaultNone, primary_keyTrue)username: str Field(indexTrue, nullableFalse, uniqueTrue)email: str Field(indexTrue, nullableFalse, uniqueTrue)hashed_password: str Field(nullableFalse)is_active: bool Field(defaultTrue)created_at: datetime Field(default_factorydatetime.utcnow)addresses: List[Address] Relationship(back_populatesuser)schemas.py
from typing import List, Optional
from pydantic import BaseModel, EmailStr
from datetime import datetimeclass AddressCreate(BaseModel):street: strcity: strclass AddressRead(BaseModel):id: intstreet: strcity: strclass Config:orm_mode Trueclass UserCreate(BaseModel):username: stremail: EmailStrpassword: strclass UserRead(BaseModel):id: intusername: stremail: EmailStris_active: boolcreated_at: datetimeaddresses: List[AddressRead] []class Config:orm_mode Truedatabase.py
from sqlmodel import SQLModel, create_engine, Session
from models import User, AddressDATABASE_URL sqlite:///./test.db
engine create_engine(DATABASE_URL, echoTrue)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionmain.py
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from models import User, Address
from schemas import UserCreate, UserRead, AddressCreate, AddressRead
from database import create_db_and_tables, get_session
from typing import Listapp FastAPI()app.on_event(startup)
def on_startup():create_db_and_tables()app.post(/users/, response_modelUserRead)
def create_user(user: UserCreate, session: Session Depends(get_session)):db_user session.exec(select(User).where(User.username user.username)).first()if db_user:raise HTTPException(status_code400, detailUsername already exists)new_user User(usernameuser.username,emailuser.email,hashed_passworduser.password # 实际项目中应进行哈希处理)session.add(new_user)session.commit()session.refresh(new_user)return new_userapp.get(/users/{user_id}, response_modelUserRead)
def read_user(user_id: int, session: Session Depends(get_session)):user session.get(User, user_id)if not user:raise HTTPException(status_code404, detailUser not found)return userapp.post(/users/{user_id}/addresses/, response_modelAddressRead)
def create_address(user_id: int, address: AddressCreate, session: Session Depends(get_session)):user session.get(User, user_id)if not user:raise HTTPException(status_code404, detailUser not found)new_address Address(**address.dict(), user_iduser_id)session.add(new_address)session.commit()session.refresh(new_address)return new_addressapp.get(/users/{user_id}/addresses/, response_modelList[AddressRead])
def read_addresses(user_id: int, session: Session Depends(get_session)):user session.get(User, user_id)if not user:raise HTTPException(status_code404, detailUser not found)return user.addresses运行应用
使用 uvicorn 运行 FastAPI 应用
uvicorn main:app --reload访问 http://127.0.0.1:8000/docs 查看自动生成的 API 文档并进行测试。
