Add tests with Cassandra/ScyllaDB containers for NoSQL tutorial #14350
-
First Check
Commit to Help
Example Code"""
The NoSQL tutorial (docs_src/nosql_databases/tutorial001_scylla.py) demonstrates
FastAPI integration with ScyllaDB:
"""
from contextlib import asynccontextmanager
from typing import List, Union
from uuid import UUID, uuid4
from cassandra.cluster import Cluster
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, Field
class TaskBase(BaseModel):
title: str
description: Union[str, None] = None
status: str = "pending"
class TaskCreate(TaskBase):
pass
class Task(TaskBase):
id: UUID = Field(default_factory=uuid4)
class ScyllaDBConnection:
def __init__(self, hosts=["scylladb"], port=9042):
self.cluster = Cluster(hosts, port=port)
self.session = None
self.keyspace = "task_manager"
def connect(self):
self.session = self.cluster.connect()
self.create_keyspace()
self.session.set_keyspace(self.keyspace)
self.create_table()
def create_keyspace(self):
self.session.execute(
f"""
CREATE KEYSPACE IF NOT EXISTS {self.keyspace}
WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': 1}}
"""
)
def create_table(self):
self.session.execute(
"""
CREATE TABLE IF NOT EXISTS tasks (
id uuid PRIMARY KEY,
title text,
description text,
status text,
created_at timestamp,
updated_at timestamp
)
"""
)
def close(self):
if self.session:
self.session.shutdown()
if self.cluster:
self.cluster.shutdown()
db = ScyllaDBConnection()
@asynccontextmanager
async def lifespan(app: FastAPI):
db.connect()
yield
db.close()
app = FastAPI(lifespan=lifespan)
def get_db():
return db.session
@app.post("/tasks/", response_model=Task)
def create_task(task: TaskCreate, session=Depends(get_db)):
task_id = uuid4()
query = """
INSERT INTO tasks (id, title, description, status, created_at, updated_at)
VALUES (%s, %s, %s, %s, toTimestamp(now()), toTimestamp(now()))
"""
session.execute(query, (task_id, task.title, task.description, task.status))
return Task(id=task_id, **task.model_dump())
@app.get("/tasks/", response_model=List[Task])
def read_tasks(session=Depends(get_db)):
query = "SELECT id, title, description, status FROM tasks"
rows = session.execute(query)
return [
Task(id=row.id, title=row.title, description=row.description, status=row.status)
for row in rows
]DescriptionFastAPI documentation currently doesn't mention NoSQL database integration, even though FastAPI works well with NoSQL databases like ScyllaDB, Cassandra, MongoDB, Redis, etc. The problem:
What's already happening in the community:
What could happen instead:
This would help FastAPI users discover NoSQL options and prove FastAPI's flexibility beyond SQL databases. Operating SystemmacOS Operating System DetailsNo response FastAPI Version0.121.1 Pydantic Version2.12.4 Python Version3.14.0 Additional ContextI have prepared a PR with a complete NoSQL tutorial for ScyllaDB/Cassandra, including documentation and code examples. The tutorial code has been manually tested locally against real ScyllaDB and Cassandra containers using Docker Compose to verify that it works correctly. Current testing approach:
This differs from the SQL tutorial which uses real SQLite (in-memory database). Suggested future enhancement:
Community validation:
I'm happy to implement integration tests in the initial PR or as a follow-up, depending on maintainer preference. |
Beta Was this translation helpful? Give feedback.
Replies: 3 comments
This comment was marked as disruptive content.
This comment was marked as disruptive content.
-
|
Hi @tiangolo - I’d really appreciate your input on this when you have some time. Thanks a lot for FastAPI - I’ve enjoyed working with it a lot. |
Beta Was this translation helpful? Give feedback.
-
|
Thanks for the interest! As I mentioned in the PR, I'm currently not using Cassandra, I wouldn't have the bandwidth to maintain this extra tutorial. So I'll pass on this one for now. Thanks! ☕ |
Beta Was this translation helpful? Give feedback.
Thanks for the interest! As I mentioned in the PR, I'm currently not using Cassandra, I wouldn't have the bandwidth to maintain this extra tutorial. So I'll pass on this one for now. Thanks! ☕