]> git.ipfire.org Git - thirdparty/starlette.git/commitdiff
Remove routing decorators in test_databases.py (#1483)
authorAmin Alaee <mohammadamin.alaee@gmail.com>
Wed, 9 Feb 2022 09:15:50 +0000 (10:15 +0100)
committerGitHub <noreply@github.com>
Wed, 9 Feb 2022 09:15:50 +0000 (09:15 +0000)
tests/test_database.py

index 11f770bb1f385a00af9b63926b34fd08a7d763b8..f3fe9c52f1ff8ac9a0dbc2f12a0f2546907acb27 100644 (file)
@@ -3,7 +3,9 @@ import pytest
 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"
 
@@ -29,22 +31,18 @@ def create_test_database():
     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 = [
@@ -53,9 +51,8 @@ async def list_notes(request):
     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)
@@ -64,16 +61,14 @@ async def add_note(request):
     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)
@@ -82,8 +77,7 @@ async def read_note(request):
     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)
@@ -91,6 +85,19 @@ async def read_note_text(request):
     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(