AirModel实战教程：基于PostgreSQL的异步ORM轻松上手-拓冰建站

AirModel实战教程：基于PostgreSQL的异步ORM轻松上手

【免费下载链接】airThe first web framework designed for AI to write. Built on Python, FastAPI, Pydantic, and HTMX. By the authors of Two Scoops of Django.项目地址: https://gitcode.com/gh_mirrors/air23/air

AirModel是GitHub加速计划（air23/air）项目中内置的异步ORM工具，专为PostgreSQL数据库设计，让开发者能够轻松实现数据模型与数据库的交互。作为首个专为AI编写的Web框架，AirModel基于Python、FastAPI、Pydantic和HTMX构建，提供了简洁高效的数据库操作体验。

为什么选择AirModel？

AirModel作为一款现代化的异步ORM，具有以下核心优势：

简单易用：只需一个导入和一个基类，就能定义数据模型并处理验证、序列化和异步数据库操作
自动迁移：添加字段后，create_tables()会自动使用ALTER TABLE ADD COLUMN迁移现有表
Django风格API：提供create、get、filter、save、delete等直观方法
类型安全：基于Pydantic构建，提供完整的类型注解和编辑器支持

快速开始：安装与配置

要使用AirModel，首先需要安装Air框架。通过以下命令克隆项目仓库：

git clone https://gitcode.com/gh_mirrors/air23/air

然后安装依赖：

cd air uv sync

配置PostgreSQL连接，只需设置环境变量：

export DATABASE_URL="postgresql://user:password@localhost/dbname?sslmode=require"

Air会在应用启动时自动打开asyncpg连接池，并在关闭时自动关闭连接池，无需手动管理连接。

定义你的第一个AirModel模型

创建数据模型非常简单，只需继承AirModel并定义字段：

from air import AirField, AirModel class UnicornSighting(AirModel): id: int = AirField(primary_key=True) name: str location: str sparkle_rating: int sighted_at: datetime

每个类型注解都会直接映射到PostgreSQL列类型：

Python类型	PostgreSQL类型
int	INTEGER
str	TEXT
bool	BOOLEAN
datetime	TIMESTAMPTZ
float	DOUBLE PRECISION

类名会自动从CamelCase转换为snake_case作为表名，符合PostgreSQL的命名习惯。

基本CRUD操作

AirModel提供了直观的API来执行常见的数据库操作：

创建记录

# 创建单条记录 sighting = await UnicornSighting.create( name="Sparkle Hoof", location="Rainbow Falls", sparkle_rating=9, sighted_at=datetime.now() ) # 批量创建记录 sightings = await UnicornSighting.bulk_create([ UnicornSighting(name="Moonbeam", location="Star Mountain", sparkle_rating=8), UnicornSighting(name="Fluffball", location="Cloud Valley", sparkle_rating=7) ])

查询记录

# 获取单条记录 sighting = await UnicornSighting.get(id=1) # 过滤记录 high_rated = await UnicornSighting.filter(sparkle_rating__gte=8) # 排序和分页 recent_sightings = await UnicornSighting.filter( location__icontains="falls" ).order_by("-sighted_at").limit(10).offset(20) # 计数 count = await UnicornSighting.count(sparkle_rating__gte=9)

更新记录

sighting = await UnicornSighting.get(id=1) sighting.sparkle_rating = 10 await sighting.save() # 批量更新 await UnicornSighting.filter(location="Rainbow Falls").update(sparkle_rating=9)

删除记录

sighting = await UnicornSighting.get(id=1) await sighting.delete() # 批量删除 await UnicornSighting.filter(sparkle_rating__lt=5).delete()

事务处理

AirModel支持事务操作，确保数据一致性：

async with app.db.transaction(): sighting1 = await UnicornSighting.create(name="Sunny", location="Meadow", sparkle_rating=8) sighting2 = await UnicornSighting.create(name="Twilight", location="Forest", sparkle_rating=9) # 如果发生异常，所有操作都会回滚

与AirForm集成

AirModel可以与AirForm无缝集成，实现表单验证和数据库操作的一体化：

from air import AirForm, AirField, AirModel class ContactMessage(AirModel): id: int = AirField(primary_key=True) name: str = AirField(..., max_length=100) email: str = AirField(..., format="email") message: str = AirField(..., widget="textarea") class ContactForm(AirForm[ContactMessage]): pass # 在请求处理中使用 async def handle_contact(request): form = await ContactForm.from_request(request) if form.is_valid(): # 直接保存到数据库 await form.data.create() return RedirectResponse("/thanks") return TemplateResponse("contact.html", {"form": form})