feat(scaffold): add model-backed CRUD service template

Provides a complete OpenAPI-first CRUD example with a Pydantic
model layer, explicit runtime semantics, and integration tests
to support developer onboarding and real-world service structure.
This commit is contained in:
2026-01-10 18:05:39 +05:30
parent 2ac342240b
commit 40d91bc52b
6 changed files with 332 additions and 0 deletions

View File

@@ -0,0 +1,44 @@
"""
In-memory data store using Pydantic models.
This module is NOT thread-safe and is intended for demos and scaffolds only.
"""
from typing import Dict
from models import Item, ItemCreate
_items: Dict[int, Item] = {
1: Item(id=1, name="Apple", price=0.5),
2: Item(id=2, name="Banana", price=0.3),
}
_next_id = 3
def list_items() -> list[Item]:
return list(_items.values())
def get_item(item_id: int) -> Item:
return _items[item_id]
def create_item(payload: ItemCreate) -> Item:
global _next_id
item = Item(id=_next_id, **payload.model_dump())
_items[_next_id] = item
_next_id += 1
return item
def update_item(item_id: int, payload: ItemCreate) -> Item:
if item_id not in _items:
raise KeyError(item_id)
item = Item(id=item_id, **payload.model_dump())
_items[item_id] = item
return item
def delete_item(item_id: int) -> None:
del _items[item_id]

View File

@@ -0,0 +1,8 @@
from openapi_first.app import OpenAPIFirstApp
import routes
app = OpenAPIFirstApp(
openapi_path="openapi.yaml",
routes_module=routes,
title="Model CRUD Example Service",
)

View File

@@ -0,0 +1,18 @@
"""
Pydantic domain models for the CRUD example.
"""
from pydantic import BaseModel
class ItemBase(BaseModel):
name: str
price: float
class ItemCreate(ItemBase):
pass
class Item(ItemBase):
id: int

View File

@@ -0,0 +1,116 @@
openapi: 3.0.3
info:
title: Model CRUD Example Service
version: "1.0.0"
description: OpenAPI-first CRUD service with Pydantic models.
paths:
/items:
get:
operationId: list_items
responses:
"200":
description: List of items
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/Item"
post:
operationId: create_item
requestBody:
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/ItemCreate"
responses:
"201":
description: Item created
content:
application/json:
schema:
$ref: "#/components/schemas/Item"
/items/{item_id}:
get:
operationId: get_item
parameters:
- name: item_id
in: path
required: true
schema:
type: integer
responses:
"200":
description: Item found
content:
application/json:
schema:
$ref: "#/components/schemas/Item"
"404":
description: Item not found
put:
operationId: update_item
parameters:
- name: item_id
in: path
required: true
schema:
type: integer
requestBody:
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/ItemCreate"
responses:
"200":
description: Item updated
content:
application/json:
schema:
$ref: "#/components/schemas/Item"
"404":
description: Item not found
delete:
operationId: delete_item
parameters:
- name: item_id
in: path
required: true
schema:
type: integer
responses:
"204":
description: Item deleted
"404":
description: Item not found
components:
schemas:
Item:
type: object
required: [id, name, price]
properties:
id:
type: integer
name:
type: string
price:
type: number
format: float
ItemCreate:
type: object
required: [name, price]
properties:
name:
type: string
price:
type: number
format: float

View File

@@ -0,0 +1,48 @@
"""
CRUD route handlers bound via OpenAPI operationId.
"""
from fastapi import Response, HTTPException
from models import ItemCreate
from data import (
list_items as _list_items,
get_item as _get_item,
create_item as _create_item,
update_item as _update_item,
delete_item as _delete_item,
)
def list_items():
return _list_items()
def get_item(item_id: int):
try:
return _get_item(item_id)
except KeyError:
raise HTTPException(status_code=404, detail="Item not found")
def create_item(payload: ItemCreate, response: Response):
item = _create_item(payload)
response.status_code = 201
return item
def update_item(item_id: int, payload: ItemCreate):
try:
return _update_item(item_id, payload)
except KeyError:
raise HTTPException(status_code=404, detail="Item not found")
def delete_item(item_id: int, response: Response):
try:
_delete_item(item_id)
except KeyError:
raise HTTPException(status_code=404, detail="Item not found")
response.status_code = 204
return None

View File

@@ -0,0 +1,98 @@
"""
End-to-end tests for the OpenAPI-first model CRUD example app.
These tests validate that all CRUD operations behave correctly
against the in-memory mock data store using Pydantic models.
The tests assume:
- OpenAPI-first route binding
- Pydantic model validation
- In-memory storage (no persistence guarantees)
- Deterministic behavior in a single process
"""
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_list_items_initial():
"""Initial items should be present."""
response = client.get("/items")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) >= 2
ids = {item["id"] for item in data}
assert 1 in ids
assert 2 in ids
def test_get_item():
"""Existing item should be retrievable by ID."""
response = client.get("/items/1")
assert response.status_code == 200
item = response.json()
assert item["id"] == 1
assert isinstance(item["name"], str)
assert isinstance(item["price"], float)
def test_create_item():
"""Creating a new item should return the created entity."""
payload = {
"name": "Orange",
"price": 0.8,
}
response = client.post("/items", json=payload)
assert response.status_code == 201
item = response.json()
assert "id" in item
assert item["name"] == payload["name"]
assert item["price"] == payload["price"]
# Verify it appears in list
list_response = client.get("/items")
ids = {i["id"] for i in list_response.json()}
assert item["id"] in ids
def test_update_item():
"""Updating an item should replace its values."""
payload = {
"name": "Green Apple",
"price": 0.6,
}
response = client.put("/items/1", json=payload)
assert response.status_code == 200
item = response.json()
assert item["id"] == 1
assert item["name"] == payload["name"]
assert item["price"] == payload["price"]
# Verify persisted update
get_response = client.get("/items/1")
updated = get_response.json()
assert updated["name"] == payload["name"]
assert updated["price"] == payload["price"]
def test_delete_item():
"""Deleting an item should remove it from the store."""
response = client.delete("/items/2")
assert response.status_code == 204
# Verify deletion
list_response = client.get("/items")
ids = {item["id"] for item in list_response.json()}
assert 2 not in ids