import sqlalchemy
from starlette.applications import Starlette
+from starlette.requests import Request
from starlette.responses import JSONResponse
+from starlette.routing import Route
DATABASE_URL = "sqlite:///test.db"
metadata.drop_all(engine)
-app = Starlette()
database = databases.Database(DATABASE_URL, force_rollback=True)
-@app.on_event("startup")
async def startup():
await database.connect()
-@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
-@app.route("/notes", methods=["GET"])
-async def list_notes(request):
+async def list_notes(request: Request):
query = notes.select()
results = await database.fetch_all(query)
content = [
return JSONResponse(content)
-@app.route("/notes", methods=["POST"])
@database.transaction()
-async def add_note(request):
+async def add_note(request: Request):
data = await request.json()
query = notes.insert().values(text=data["text"], completed=data["completed"])
await database.execute(query)
return JSONResponse({"text": data["text"], "completed": data["completed"]})
-@app.route("/notes/bulk_create", methods=["POST"])
-async def bulk_create_notes(request):
+async def bulk_create_notes(request: Request):
data = await request.json()
query = notes.insert()
await database.execute_many(query, data)
return JSONResponse({"notes": data})
-@app.route("/notes/{note_id:int}", methods=["GET"])
-async def read_note(request):
+async def read_note(request: Request):
note_id = request.path_params["note_id"]
query = notes.select().where(notes.c.id == note_id)
result = await database.fetch_one(query)
return JSONResponse(content)
-@app.route("/notes/{note_id:int}/text", methods=["GET"])
-async def read_note_text(request):
+async def read_note_text(request: Request):
note_id = request.path_params["note_id"]
query = sqlalchemy.select([notes.c.text]).where(notes.c.id == note_id)
result = await database.fetch_one(query)
return JSONResponse(result[0])
+app = Starlette(
+ routes=[
+ Route("/notes", endpoint=list_notes, methods=["GET"]),
+ Route("/notes", endpoint=add_note, methods=["POST"]),
+ Route("/notes/bulk_create", endpoint=bulk_create_notes, methods=["POST"]),
+ Route("/notes/{note_id:int}", endpoint=read_note, methods=["GET"]),
+ Route("/notes/{note_id:int}/text", endpoint=read_note_text, methods=["GET"]),
+ ],
+ on_startup=[startup],
+ on_shutdown=[shutdown],
+)
+
+
def test_database(test_client_factory):
with test_client_factory(app) as client:
response = client.post(