首先

有关Apache Cassandra

Apache Cassandra是一个开源的分布式数据库管理系统。

与其他分布式数据库管理系统类似，我们使用多个通用服务器来构建一个数据库（只使用一个服务器进行构建也是可能的，根据开发等需求来决定）。

在这里，我们将省略详细的解释，将介绍的责任留给对此有兴趣的人，官方网站和维基百科会提供更详细的介绍。

使用Cassandra + Python + FastAPI进行REST API开发。

在这篇文章中，我们将开发一个自定义的REST API，用于覆盖开发人员定义的对表进行CRUD操作。

代码也在这里公开。

我已经用英文在上述存储库中写了说明，但在这里我将用日文来解释。

引言：关于DataStax Astra DB

在这个程序中，我们将使用DataStax提供的DBaaS服务AstraDB作为连接目标的Cassandra数据库。

使用DataStax AstraDB的开发人员可以根据AstraDB控制平面（数据库仪表板的连接选项卡）提供的具体步骤，轻松确认从各种编程语言的连接。使用的示例程序用于连接到数据库并显示系统表（system.local）的信息，用于连接确认（例如，如Python中的connect_database.py）。

此外，AstraDB默认提供了用于对Cassandra表执行CRUD操作的REST API。因此，这里提供的代码可能看起来像是重新发明轮子。然而，在实际项目中，并不总是将数据操作本身作为API提供和使用的情况，往往涉及一些业务逻辑。例如，一个API调用可能包括对多个表的操作。在这里提供的代码虽然没有实现特定的逻辑，但可以考虑它作为模拟带有业务逻辑的API，或者作为开发这种API的样板。

开发的物品

样例程序提供REST API。
此外，还提供符合Open API标准的Web UI。

开发过程

准备好

在这里，我们假设在AstraDB上已经创建了数据库。

另外，在 AstraDB 的控制平面（数据库仪表板的「连接」选项卡中）提供的步骤，假设您已经熟悉。

请放心，上述准备工作应该并不复杂。您不需要输入信用卡信息即可完成注册。请务必试用一下。

另外，如果将数据库连接作为单独的部分，也可以将其应用到开源软件Apache Cassandra上。

Python的應用环境

准备Python环境。这里将展示在Windows上操作的记录。

conda create -n datastax -y python=3 pip
conda activate datastax

python --version
Python 3.10.4
pip --version
pip 21.2.4 from C:\Users\yoshi\Anaconda3\envs\datastax\lib\site-packages\pip (python 3.10)

Python软件包

安装所需的软件包。

pip install fastapi
pip install uvicorn[standard]

pip install cassandra-driver

数据模型

我将在Cassandra（AstraDB）中创建以下的键空间和表。

键空间: 测试

表格：会员

CREATE TABLE test.member (
    id text PRIMARY KEY,
    first_name text,
    last_name text
) 

代码 ma)

请按照AstraDB控制平面（数据库仪表板的”连接”选项卡）提供的指示，对以下代码（astra_main.py）进行编辑。

from fastapi import FastAPI
from pydantic import BaseModel

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider

cloud_config= {
        'secure_connect_bundle': './exclude/secure-connect-<<YOUR DB NAME>>.zip'
}
auth_provider = PlainTextAuthProvider('<<CLIENT ID>>', '<<CLIENT SECRET>>')
cluster = Cluster(cloud=cloud_config, auth_provider=auth_provider)
session = cluster.connect()

app = FastAPI()

class Member(BaseModel):
    first_name: str
    last_name: str
    id: str

@app.get("/")
async def read_root():
    return {"Status": "On"}


@app.get("/members/{member_id}")
async def read_member(member_id: str):

    row = session.execute("select first_name, last_name from test.member where id = %s",[member_id]).one()

    if row:
        return {"member_id": member_id, "first_name":row[0], "last_name": row[1]}
    else:
        return {"member_id": member_id}

@app.put("/members/upsert/{member_id}")
def upsert_member(member_id: str, member: Member):

    session.execute("""
        insert into test.member (id, first_name, last_name)
        values (%s, %s, %s)
        """,
        (member_id, member.first_name, member.last_name)
    )
    return {"member_id": member_id}

@app.get("/members/delete/{member_id}")
async def delete_member(member_id: str):

    row = session.execute("delete from test.member where id = %s",[member_id]).one()

    return {"deleted": member_id}

利用法 – 借助方法

请使用以下命令进行执行。

uvicorn astra_main:app --reload

我会在下面的链接中进行确认。

最后

我已经了解了使用Cassandra和Python组合创建REST API的方法。
关于FastAPI的详细信息，请查阅以下文档等。

希望您能通过查看我们介绍的样例代码来了解与Cassandra的CRUD操作相关的编码实现。