mirror
/
tiangolo.fastapi
mirror of https://github.com/tiangolo/fastapi
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

[14350] Added ScyllaDB and Cassandra docs and tests

pull/14352/head
Andrii Kysylevskyi 8 months ago
parent
02e108d166
commit
fe690664a3
6 changed files with 1062 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 341
      docs/en/docs/tutorial/nosql-databases.md
  2. 143
      docs_src/nosql_databases/tutorial001.py
  3. 143
      docs_src/nosql_databases/tutorial001_scylla.py
  4. 0
      tests/test_tutorial/test_nosql_databases/__init__.py
  5. 12
      tests/test_tutorial/test_nosql_databases/conftest.py
  6. 423
      tests/test_tutorial/test_nosql_databases/test_tutorial001.py

341
docs/en/docs/tutorial/nosql-databases.md
View File

@ -0,0 +1,341 @@
# NoSQL Databases { #nosql-databases }
**FastAPI** doesn't require you to use a SQL (relational) database.
Here we'll see an example using <a href="https://cassandra.apache.org/" class="external-link" target="_blank">Apache Cassandra</a>, a popular distributed NoSQL database.
We'll also show how <a href="https://www.scylladb.com/" class="external-link" target="_blank">ScyllaDB</a>, a Cassandra-compatible database, works using the exact same code.
/// tip
FastAPI doesn't force you to use any specific database. This tutorial demonstrates Cassandra/ScyllaDB, but you can use other databases with their respective Python drivers.
///
## Install Dependencies { #install-dependencies }
First, make sure you create your [virtual environment](../virtual-environments.md){.internal-link target=_blank}, activate it, and then install the Cassandra driver:
<div class="termy">
```console
$ pip install cassandra-driver
---> 100%
```
</div>
/// note
The same `cassandra-driver` works with both Apache Cassandra and ScyllaDB. This is what makes ScyllaDB a true drop-in replacement.
///
/// tip | Python 3.12+
If you're using Python 3.12 or newer, you'll also need an event loop implementation. Install `gevent` for best compatibility:
<div class="termy">
```console
$ pip install gevent
---> 100%
```
</div>
The `asyncore` module was removed in Python 3.12, so `cassandra-driver` requires an alternative event loop like gevent or libev.
///
## Set Up Docker Containers { #set-up-docker-containers }
For local development and testing, you can use Docker Compose to run both Cassandra and ScyllaDB.
Create a `docker-compose.yml` file:
```yaml
services:
cassandra:
image: cassandra:4.1
container_name: fastapi-cassandra
ports:
- "9043:9042"
environment:
- CASSANDRA_CLUSTER_NAME=test-cluster
healthcheck:
test: ["CMD-SHELL", "cqlsh -e 'DESCRIBE KEYSPACES'"]
interval: 10s
timeout: 5s
retries: 10
volumes:
- cassandra_data:/var/lib/cassandra
scylladb:
image: scylladb/scylla:2025.1.4
container_name: fastapi-scylladb
command: --reactor-backend epoll --smp 1 --memory 1G --overprovisioned 1 --api-address 0.0.0.0
ports:
- "9042:9042"
healthcheck:
test: ["CMD-SHELL", "cqlsh -e 'DESCRIBE KEYSPACES'"]
interval: 10s
timeout: 5s
retries: 10
volumes:
- scylladb_data:/var/lib/scylla
volumes:
cassandra_data:
scylladb_data:
```
Start the containers:
<div class="termy">
```console
$ docker-compose up -d
```
</div>
/// tip
Notice that both containers expose port 9042 (the CQL native protocol port). We map them to different host ports (9042 and 9043) so both can run simultaneously for comparison.
///
## Create the App { #create-the-app }
We'll create a task management API that demonstrates CRUD operations with Cassandra.
### Import and Create Models { #import-and-create-models }
We use **Pydantic models** for data validation, just like with SQL databases:
{* ../../docs_src/nosql_databases/tutorial001.py ln[1:17] hl[1:5,8:17] *}
/// tip
Unlike SQL databases, Cassandra uses UUIDs for primary keys. This works great in distributed systems because UUIDs can be generated independently on any node without conflicts.
///
### Create the Database Connection { #create-the-database-connection }
Create a connection class that manages the Cassandra cluster, session, and schema:
{* ../../docs_src/nosql_databases/tutorial001.py ln[20:59] hl[20:22,24:26,28:35,37:51] *}
Notice:
* **Cluster** connects to Cassandra nodes (line 21)
* **Session** executes CQL queries (line 25)
* **Keyspace** is like a database/schema in SQL (line 26)
* **Replication** strategy determines how data is distributed
/// note
`SimpleStrategy` with `replication_factor: 1` is for development. In production, use `NetworkTopologyStrategy` with multiple replicas across data centers.
///
### Initialize the Database { #initialize-the-database }
Set up the application and database connection using the modern `lifespan` context manager:
{* ../../docs_src/nosql_databases/tutorial001.py ln[65:81] hl[65,68:74,77] *}
/// tip
We use the `lifespan` context manager to handle startup and shutdown events. This is the recommended approach in modern FastAPI applications. See the <a href="https://fastapi.tiangolo.com/advanced/events/" class="internal-link" target="_blank">events documentation</a> for more details.
///
### Create a Task { #create-a-task }
Add a **POST** endpoint to create tasks:
{* ../../docs_src/nosql_databases/tutorial001.py ln[84:93] hl[84,85,86:92] *}
Key points:
* Generate UUID for primary key
* Use CQL parameterized queries (secure against injection)
* Use `toTimestamp(now())` for Cassandra timestamps
### Read Tasks { #read-tasks }
Add a **GET** endpoint to list all tasks:
{* ../../docs_src/nosql_databases/tutorial001.py ln[96:104] hl[96,97,98:99,100:103] *}
### Read One Task { #read-one-task }
Add a **GET** endpoint for a specific task:
{* ../../docs_src/nosql_databases/tutorial001.py ln[107:112] hl[107,108,109:110,111] *}
### Update a Task { #update-a-task }
Add a **PUT** endpoint to update tasks:
{* ../../docs_src/nosql_databases/tutorial001.py ln[115:131] hl[115,116,117:121,123:127] *}
### Delete a Task { #delete-a-task }
Add a **DELETE** endpoint:
{* ../../docs_src/nosql_databases/tutorial001.py ln[134:144] hl[134,135,136:140,142:143] *}
## Run the App { #run-the-app }
Save the code to `main.py` and run it:
<div class="termy">
```console
$ fastapi dev main.py
╭────────── FastAPI CLI - Development mode ───────────╮
│ │
│ Serving at: http://127.0.0.1:8000 │
│ │
│ API docs: http://127.0.0.1:8000/docs │
│ │
│ Running in development mode, for production use: │
│ │
│ fastapi run │
│ │
╰─────────────────────────────────────────────────────╯
INFO: Will watch for changes in these directories: ['/home/user/code']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [2248755] using WatchFiles
INFO: Started server process [2248757]
INFO: Waiting for application startup.
INFO: Application startup complete.
```
</div>
## Check the API Docs { #check-the-api-docs }
Open your browser at <a href="http://127.0.0.1:8000/docs" class="external-link" target="_blank">http://127.0.0.1:8000/docs</a>.
You will see the automatic interactive API documentation (provided by Swagger UI):
<img src="/img/tutorial/nosql-databases/image01.png">
## Use ScyllaDB Instead { #use-scylladb-instead }
**ScyllaDB is Cassandra-compatible**, so you can use it with the same code.
To use ScyllaDB instead of Cassandra, change the **hostname** in the connection:
```python
# Cassandra version (tutorial001.py)
class CassandraConnection:
def __init__(self, hosts=["cassandra"], port=9042):
# ... rest of the code stays the same
```
Change to:
```python
# ScyllaDB version (tutorial001_scylla.py)
class ScyllaDBConnection:
def __init__(self, hosts=["scylladb"], port=9042):
# ... rest of the code stays the same
```
That's it! Everything else is **identical**:
* ✅ Same Cassandra driver (`cassandra-driver`)
* ✅ Same CQL queries
* ✅ Same data models
* ✅ Same API endpoints
* ✅ Same behavior
The complete ScyllaDB version is in `docs_src/nosql_databases/tutorial001_scylla.py` - the only difference is the hostname.
## Production Considerations { #production-considerations }
### Connection Pooling { #connection-pooling }
The Cassandra driver automatically manages connection pools. For production, configure:
```python
from cassandra.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT
from cassandra.policies import DCAwareRoundRobinPolicy, TokenAwarePolicy
profile = ExecutionProfile(
load_balancing_policy=TokenAwarePolicy(DCAwareRoundRobinPolicy()),
request_timeout=15
)
cluster = Cluster(
hosts=['node1', 'node2', 'node3'],
execution_profiles={EXEC_PROFILE_DEFAULT: profile}
)
```
### Consistency Levels { #consistency-levels }
Cassandra allows tuning consistency per query:
```python
from cassandra.query import SimpleStatement
from cassandra import ConsistencyLevel
query = SimpleStatement(
"SELECT * FROM tasks WHERE id = %s",
consistency_level=ConsistencyLevel.QUORUM
)
session.execute(query, (task_id,))
```
### Error Handling { #error-handling }
Add proper error handling for production:
```python
from cassandra.cluster import NoHostAvailable
from cassandra import OperationTimedOut
try:
session.execute(query)
except NoHostAvailable:
raise HTTPException(status_code=503, detail="Database unavailable")
except OperationTimedOut:
raise HTTPException(status_code=504, detail="Query timeout")
```
### Schema Migrations { #schema-migrations }
For production, use migration tools:
* <a href="https://github.com/Cobliteam/cassandra-migrate" class="external-link" target="_blank">cassandra-migrate</a>
* Custom CQL scripts with version tracking
* Application-level schema management
## Learn More { #learn-more }
This is a quick introduction. For more advanced topics, see:
* <a href="https://cassandra.apache.org/doc/latest/" class="external-link" target="_blank">Cassandra Documentation</a>
* <a href="https://opensource.docs.scylladb.com/" class="external-link" target="_blank">ScyllaDB Documentation</a>
* <a href="https://docs.datastax.com/en/developer/python-driver/" class="external-link" target="_blank">DataStax Python Driver Documentation</a>
## Recap { #recap }
FastAPI works great with NoSQL databases like Cassandra and ScyllaDB:
* ✅ Use standard Python drivers
* ✅ Leverage Pydantic for data validation
* ✅ Get automatic API documentation
* ✅ Enjoy type safety and editor support
* ✅ Switch between compatible databases with minimal changes
FastAPI works well with NoSQL databases, providing the same benefits as with SQL databases.

143
docs_src/nosql_databases/tutorial001.py
View File

@ -0,0 +1,143 @@
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 CassandraConnection:
def __init__(self, hosts=["cassandra"], 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 = CassandraConnection()
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
db.connect()
yield
# Shutdown
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
]
@app.get("/tasks/{task_id}", response_model=Task)
def read_task(task_id: UUID, session=Depends(get_db)):
query = "SELECT id, title, description, status FROM tasks WHERE id = %s"
row = session.execute(query, (task_id,)).one()
if not row:
raise HTTPException(status_code=404, detail="Task not found")
return Task(id=row.id, title=row.title, description=row.description, status=row.status)
@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: UUID, task: TaskCreate, session=Depends(get_db)):
# Check if task exists
check_query = "SELECT id FROM tasks WHERE id = %s"
existing = session.execute(check_query, (task_id,)).one()
if not existing:
raise HTTPException(status_code=404, detail="Task not found")
update_query = """
UPDATE tasks
SET title = %s, description = %s, status = %s, updated_at = toTimestamp(now())
WHERE id = %s
"""
session.execute(
update_query, (task.title, task.description, task.status, task_id)
)
return Task(id=task_id, **task.model_dump())
@app.delete("/tasks/{task_id}")
def delete_task(task_id: UUID, session=Depends(get_db)):
# Check if task exists
check_query = "SELECT id FROM tasks WHERE id = %s"
existing = session.execute(check_query, (task_id,)).one()
if not existing:
raise HTTPException(status_code=404, detail="Task not found")
delete_query = "DELETE FROM tasks WHERE id = %s"
session.execute(delete_query, (task_id,))
return {"ok": True}

143
docs_src/nosql_databases/tutorial001_scylla.py
View File

@ -0,0 +1,143 @@
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):
# Startup
db.connect()
yield
# Shutdown
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
]
@app.get("/tasks/{task_id}", response_model=Task)
def read_task(task_id: UUID, session=Depends(get_db)):
query = "SELECT id, title, description, status FROM tasks WHERE id = %s"
row = session.execute(query, (task_id,)).one()
if not row:
raise HTTPException(status_code=404, detail="Task not found")
return Task(id=row.id, title=row.title, description=row.description, status=row.status)
@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: UUID, task: TaskCreate, session=Depends(get_db)):
# Check if task exists
check_query = "SELECT id FROM tasks WHERE id = %s"
existing = session.execute(check_query, (task_id,)).one()
if not existing:
raise HTTPException(status_code=404, detail="Task not found")
update_query = """
UPDATE tasks
SET title = %s, description = %s, status = %s, updated_at = toTimestamp(now())
WHERE id = %s
"""
session.execute(
update_query, (task.title, task.description, task.status, task_id)
)
return Task(id=task_id, **task.model_dump())
@app.delete("/tasks/{task_id}")
def delete_task(task_id: UUID, session=Depends(get_db)):
# Check if task exists
check_query = "SELECT id FROM tasks WHERE id = %s"
existing = session.execute(check_query, (task_id,)).one()
if not existing:
raise HTTPException(status_code=404, detail="Task not found")
delete_query = "DELETE FROM tasks WHERE id = %s"
session.execute(delete_query, (task_id,))
return {"ok": True}

0
tests/test_tutorial/test_nosql_databases/__init__.py
View File

12
tests/test_tutorial/test_nosql_databases/conftest.py
View File

@ -0,0 +1,12 @@
# Mock cassandra module to avoid Python 3.14+ import issues
# (asyncore removed in 3.12, cassandra-driver needs event loop)
import sys
from types import ModuleType
from unittest.mock import MagicMock
mock_cassandra = ModuleType('cassandra')
mock_cassandra_cluster = ModuleType('cassandra.cluster')
mock_cassandra_cluster.Cluster = MagicMock
sys.modules['cassandra'] = mock_cassandra
sys.modules['cassandra.cluster'] = mock_cassandra_cluster

423
tests/test_tutorial/test_nosql_databases/test_tutorial001.py
View File

@ -0,0 +1,423 @@
import importlib
from typing import Any
from unittest.mock import MagicMock, patch
from uuid import UUID
import pytest
from dirty_equals import IsDict, IsStr, IsUUID
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
class MockRow:
def __init__(self, id: UUID, title: str, description: str, status: str):
self.id = id
self.title = title
self.description = description
self.status = status
class MockResult:
def __init__(self, rows: list[MockRow]):
self._rows = rows
self._iter = iter(rows)
def __iter__(self):
return self._iter
def one(self):
if self._rows:
return self._rows[0]
return None
@pytest.fixture(
name="client",
params=["tutorial001", "tutorial001_scylla"],
)
def get_client(request: pytest.FixtureRequest):
mock_session = MagicMock()
mock_cluster_instance = MagicMock()
mock_cluster_instance.connect.return_value = mock_session
tasks_store: dict[UUID, dict[str, Any]] = {}
with patch("cassandra.cluster.Cluster") as mock_cluster:
mock_cluster.return_value = mock_cluster_instance
def mock_execute(query: str, params: tuple = ()):
if "DELETE FROM tasks" in query:
task_id = params[0]
if task_id in tasks_store:
del tasks_store[task_id]
return None
if "UPDATE tasks" in query:
title, description, status, task_id = params
if task_id in tasks_store:
tasks_store[task_id].update(
{"title": title, "description": description, "status": status}
)
return None
return None
if "INSERT INTO tasks" in query:
task_id, title, description, status = params
tasks_store[task_id] = {
"id": task_id,
"title": title,
"description": description,
"status": status,
}
return None
if "SELECT" in query and "WHERE" not in query:
rows = [
MockRow(
id=task["id"],
title=task["title"],
description=task["description"],
status=task["status"],
)
for task in tasks_store.values()
]
return MockResult(rows)
if "SELECT" in query and "WHERE id = %s" in query:
task_id = params[0]
if task_id in tasks_store:
task = tasks_store[task_id]
return MockResult(
[
MockRow(
id=task["id"],
title=task["title"],
description=task["description"],
status=task["status"],
)
]
)
return MockResult([])
return None
mock_session.execute.side_effect = mock_execute
mod = importlib.import_module(f"docs_src.nosql_databases.{request.param}")
importlib.reload(mod)
with TestClient(mod.app) as c:
yield c
def test_crud_app(client: TestClient):
response = client.get("/tasks/")
assert response.status_code == 200, response.text
assert response.json() == []
response = client.post(
"/tasks/",
json={
"title": "Buy groceries",
"description": "Milk, eggs, bread",
"status": "pending",
},
)
assert response.status_code == 200, response.text
data = response.json()
assert "id" in data
assert data["title"] == "Buy groceries"
assert data["description"] == "Milk, eggs, bread"
assert data["status"] == "pending"
task_id = data["id"]
response = client.get(f"/tasks/{task_id}")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"id": IsUUID(4),
"title": "Buy groceries",
"description": "Milk, eggs, bread",
"status": "pending",
}
)
response = client.post(
"/tasks/",
json={"title": "Walk the dog", "description": "In the park", "status": "pending"},
)
assert response.status_code == 200, response.text
response = client.post(
"/tasks/",
json={"title": "Write code", "description": None, "status": "in_progress"},
)
assert response.status_code == 200, response.text
response = client.get("/tasks/")
assert response.status_code == 200, response.text
assert len(response.json()) == 3
response = client.put(
f"/tasks/{task_id}",
json={
"title": "Buy groceries (Updated)",
"description": "Milk, eggs, bread, cheese",
"status": "completed",
},
)
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"id": IsUUID(4),
"title": "Buy groceries (Updated)",
"description": "Milk, eggs, bread, cheese",
"status": "completed",
}
)
response = client.delete(f"/tasks/{task_id}")
assert response.status_code == 200, response.text
assert response.json() == snapshot({"ok": True})
response = client.get(f"/tasks/{task_id}")
assert response.status_code == 404, response.text
response = client.delete(f"/tasks/{task_id}")
assert response.status_code == 404, response.text
assert response.json() == snapshot({"detail": "Task not found"})
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/tasks/": {
"post": {
"summary": "Create Task",
"operationId": "create_task_tasks__post",
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/TaskCreate"}
}
},
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Task"}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
"get": {
"summary": "Read Tasks",
"operationId": "read_tasks_tasks__get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {"$ref": "#/components/schemas/Task"},
"title": "Response Read Tasks Tasks Get",
}
}
},
}
},
},
},
"/tasks/{task_id}": {
"get": {
"summary": "Read Task",
"operationId": "read_task_tasks__task_id__get",
"parameters": [
{
"name": "task_id",
"in": "path",
"required": True,
"schema": {"type": "string", "format": "uuid", "title": "Task Id"},
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Task"}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
"put": {
"summary": "Update Task",
"operationId": "update_task_tasks__task_id__put",
"parameters": [
{
"name": "task_id",
"in": "path",
"required": True,
"schema": {"type": "string", "format": "uuid", "title": "Task Id"},
}
],
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/TaskCreate"}
}
},
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Task"}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
"delete": {
"summary": "Delete Task",
"operationId": "delete_task_tasks__task_id__delete",
"parameters": [
{
"name": "task_id",
"in": "path",
"required": True,
"schema": {"type": "string", "format": "uuid", "title": "Task Id"},
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
},
},
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {"$ref": "#/components/schemas/ValidationError"},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"Task": {
"properties": {
"title": {"type": "string", "title": "Title"},
"description": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "Description",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{"type": "string", "title": "Description"}
),
"status": {"type": "string", "default": "pending", "title": "Status"},
"id": {"type": "string", "format": "uuid", "title": "Id"},
},
"type": "object",
"required": ["title"],
"title": "Task",
},
"TaskCreate": {
"properties": {
"title": {"type": "string", "title": "Title"},
"description": IsDict(
{
"anyOf": [{"type": "string"}, {"type": "null"}],
"title": "Description",
}
)
| IsDict(
# TODO: remove when deprecating Pydantic v1
{"type": "string", "title": "Description"}
),
"status": {"type": "string", "default": "pending", "title": "Status"},
},
"type": "object",
"required": ["title"],
"title": "TaskCreate",
},
"ValidationError": {
"properties": {
"loc": {
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)
Write Preview
Loading…
Cancel
Save