在 Python 上使用IRIS REST API 进行 SQL 迁移

对于即将到来的Python 竞赛，我想制作一个小型演示，介绍如何使用 Python 创建一个简单的 REST 应用程序，该应用程序将使用 IRIS 作为数据库。使用这个工具

• FastAPI框架，高性能，易学，快速编码，可用于生产
• SQLAlchemy 是 Python SQL 工具包和对象关系映射器，为应用程序开发人员提供 SQL 的全部功能和灵活性
• Alembic 是一个轻量级数据库迁移工具，可与 SQLAlchemy Database Toolkit for Python 一起使用。
• Uvicorn 是 Python 的 ASGI Web 服务器实现。

准备环境

假设已经安装了Python，至少是3.7版本。创建一个项目文件夹，并在其中创建一个包含以下内容的requirements.txt文件

fastapi== 0.101 .1 alembic== 1.11 .1 uvicorn== 0.22 .0 sqlalchemy== 2.0 .20 sqlalchemy-iris== 0.10 .5

我建议在 python 中使用虚拟环境，让我们创建新环境并激活它

python -m venv env && source env/bin/activate

现在我们可以安装我们的依赖项

pip install -r requirements.txt

快速启动

让我们使用 FastAPI 创建最简单的 REST Api。为此，请创建app/main.py

 from fastapi import FastAPI app = FastAPI( title= 'TODO Application' , version= '1.0.0' , ) @app.get("/ping") async def pong () : return { "ping" : "pong!" }

此时就足以启动我们的应用程序，并且它应该已经可以工作了。要启动服务器，我们将使用uvicorn

-----0-----

我们可以发出ping请求

-----1-----

FastAPI 提供 UI，我们可以在其中测试 API。

Docker化环境

要将 IRIS 添加到我们的应用程序中，我们将使用容器。 IRIS 镜像将按原样运行，但我们需要为 python 应用程序构建 Docker 镜像。我们需要Dockerfile

 FROM python: 3.11 -slim-buster WORKDIR /usr/src/app RUN --mount= type = bind ,src=.,dst=. \ pip install --upgrade pip && \ pip install -r requirements.txt COPY . . ENTRYPOINT [ "/usr/src/app/entrypoint.sh" ]

要在容器内启动应用程序需要一个简单的entrypoint.sh

 #!/bin/sh # Run SQL Migrations, to make DB Schema up to date alembic upgrade head # Start Python application uvicorn app.main:app \ --workers 1 \ --host 0.0.0.0 \ --port 8000 " $@ "

不要忘记添加执行标志

chmod +x entrypoint.sh

并在docker-compose.yml中与 IRIS 结合。

 version: "3" services: iris: image: intersystemsdc/iris-community ports: - 1972 environment: - IRISUSERNAME=demo - IRISPASSWORD=demo healthcheck: test: /irisHealth.sh interval: 5 s app: build: . ports: - 8000 :8000 environment: - DATABASE_URL=iris://demo:demo@iris:1972/USER volumes: - ./:/usr/src/app depends_on: iris: condition: service_healthy command: - --reload

让我们来构建它

docker-compose build

第一个数据模型

现在让我们向应用程序声明对 IRIS 数据库的访问权限，添加文件app/db.py ，这将配置 SQLAlchemy 来访问我们的数据库，该数据库是通过 docker-compose.yml 传递的 URL 定义的。它包含几个处理程序，我们稍后将在应用程序中使用

import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base from sqlalchemy.orm import sessionmaker DATABASE_URL = os.environ.get( "DATABASE_URL" ) if not DATABASE_URL: DATABASE_URL = "iris://demo:demo@localhost:1972/USER" engine = create_engine(DATABASE_URL, echo= True , future= True ) Base: DeclarativeMeta = declarative_base() SessionLocal = sessionmaker(autocommit= False , autoflush= False , bind=engine) def init_db () : engine.connect() def get_session () : session = SessionLocal() yield session

 准备好定义我们应用程序中的第一个也是唯一一个模型。创建并编辑文件app/models.py ，它将使用 SQLAlchemy 定义模型，名为Todo ，包含三列： id、title 和 description 。

 from sqlalchemy import Column, Integer, String, Text from app.db import Base class Todo (Base) : __tablename__ = 'todo' id = Column(Integer, primary_key= True , index= True ) title = Column(String( 200 ), index= True , nullable= False ) description = Column(Text, nullable= False )

准备 SQL 迁移

在不断变化的世界中，我们知道我们的应用程序将来会得到改进，我们的表结构不是最终的，我们可以添加更多的表、列、索引等。在这种情况下，最好的方案是使用一些 SQL 迁移工具，它们有助于根据我们应用程序的版本升级数据库中的实际结构，并且使用这些工具，也有助于降级它，以防出现问题。虽然在这个项目中我们使用 Python 和 SQLAlchemy，但 SQLAlchemy 的作者提供了他的名为Alembic的工具，我们将在这里使用它。

我们需要使用我们的应用程序启动 IRIS 和容器，此时我们需要 bash，以便能够运行命令

-----2-----

运行命令alembic init app/migrations

-----3-----

这准备好了 alembic 配置，我们需要修复它以满足我们的应用程序需求。为此，请编辑app/migrations/env.py文件。这只是文件的开头，应该更新，重点更新sqlalchemy.url和target_metadata 。下面的内容保持不变

import os import urllib.parse from logging.config import fileConfig from sqlalchemy import engine_from_config from sqlalchemy import pool from alembic import context # this is the Alembic Config object, which provides # access to the values within the .ini file in use. config = context.config DATABASE_URL = os.environ.get( "DATABASE_URL" ) decoded_uri = urllib.parse.unquote(DATABASE_URL) config.set_main_option( "sqlalchemy.url" , decoded_uri) # Interpret the config file for Python logging. # This line sets up loggers basically. if config.config_file_name is not None : fileConfig(config.config_file_name) # add your model's MetaData object here # for 'autogenerate' support from app.models import Base target_metadata = Base.metadata # target_metadata = None

我们已经有了一个模型，现在我们需要使用命令alembic revision --autogenerate创建一个迁移

-----4-----

 """empty message Revision ID: 1e4d3b4d51ca Revises: Create Date: 2023-08-22 07:08:01.586330 """ from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision = '1e4d3b4d51ca' down_revision = None branch_labels = None depends_on = None def upgrade () -> None : # ### commands auto generated by Alembic - please adjust! ### op.create_table( 'todo' , sa.Column( 'id' , sa.Integer(), nullable= False ), sa.Column( 'title' , sa.String(length= 200 ), nullable= False ), sa.Column( 'description' , sa.Text(), nullable= False ), sa.PrimaryKeyConstraint( 'id' ) ) op.create_index(op.f( 'ix_todo_id' ), 'todo' , [ 'id' ], unique= False ) op.create_index(op.f( 'ix_todo_title' ), 'todo' , [ 'title' ], unique= False ) # ### end Alembic commands ### def downgrade () -> None : # ### commands auto generated by Alembic - please adjust! ### op.drop_index(op.f( 'ix_todo_title' ), table_name= 'todo' ) op.drop_index(op.f( 'ix_todo_id' ), table_name= 'todo' ) op.drop_table( 'todo' ) # ### end Alembic commands ###

现在是时候将其应用到数据库了，使用命令alembic Upgrade head ，其中 head 是升级到最新版本的关键字

-----5-----

随时检查当前状态，如果缺少某些迁移将给出信息

-----7-----

使数据可访问

所以，我们现在可以回到 REST，我们需要它启动并运行，从当前容器退出并正常运行应用程序服务，uvicorn 有一个标志 --reload，所以，它将检查 python 中的更改文件，并在我们更改它们时重新启动

-----8-----

FastAPI 使用 pydantic 项目来声明数据模式，我们也需要它，让我们创建app/schemas.py ，与 models.py 中的列相同，但采用简单的 Python 形式

from pydantic import BaseModel class TodoCreate (BaseModel) : title: str description: str class Todo (TodoCreate) : id: int class Config : from_attributes = True

在app/crud.py中声明增删改查操作，我们在其中使用 SQLAlchemy ORM 处理数据库

from sqlalchemy.orm import Session from . import models, schemas def get_todos (db: Session, skip: int = 0 , limit: int = 100 ) : return db.query(models.Todo).offset(skip).limit(limit).all() def create_todo (db: Session, todo: schemas.TodoCreate) : db_todo = models.Todo(**todo.dict()) db.add(db_todo) db.commit() db.refresh(db_todo) return db_todo

最后，我们可以更新app/main.py ，并添加读取和创建待办事项的路由

from fastapi import FastAPI, Depends from .db import init_db, get_session from . import crud, schemas app = FastAPI( title= 'TODO Application' , version= '1.0.0' , ) @app.on_event("startup") def on_startup () : init_db() @app.get("/ping") async def pong () : return { "ping" : "pong!" } @app.get("/todo", response_model=list[schemas.Todo]) async def read_todos (skip: int = 0 , limit: int = 100 , session=Depends (get_session) ) : todos = crud.get_todos(session, skip=skip, limit=limit) return todos @app.post("/todo", response_model=schemas.Todo) async def create_todo (todo: schemas.TodoCreate, session=Depends (get_session) ) : return crud.create_todo(db=session, todo=todo)

文档页面已相应更新，现在我们可以使用它了

让我们在 IRIS 中检查一下

-----9-----

希望您在创建 REST 时享受使用 Python 和 FastAPI 的轻松体验。该项目的源代码可在 github 上找到：https://github.com/caretdev/fastapi-iris-demo

原贴作者：@ Dmitry Maslennikov