-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathfastapi_integration.py
More file actions
135 lines (95 loc) · 3.87 KB
/
fastapi_integration.py
File metadata and controls
135 lines (95 loc) · 3.87 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
"""
Example: FastAPI Integration with CQRS
This example demonstrates how to integrate CQRS request handlers with FastAPI.
The system shows how to build REST APIs where FastAPI routes delegate to CQRS
handlers for business logic processing.
Use case: Building RESTful APIs with CQRS architecture. FastAPI handles HTTP
requests/responses, while CQRS handles business logic through command/query handlers.
This separation allows for better testability and maintainability.
================================================================================
HOW TO RUN THIS EXAMPLE
================================================================================
Run the server:
python examples/fastapi_integration.py
The server will start on http://localhost:8000
Option 1: Use Swagger UI
------------------------
1. Open your browser and go to:
http://localhost:8000/docs
2. Find the GET /api/hello endpoint and click "Try it out"
3. Optionally modify the "msg" query parameter
4. Click "Execute" to see the response
Option 2: Use curl
-------------------
curl "http://localhost:8000/api/hello?msg=Hello%20World"
Option 3: Use Python requests
------------------------------
import requests
response = requests.get("http://localhost:8000/api/hello", params={"msg": "Hello"})
print(response.json())
================================================================================
WHAT THIS EXAMPLE DEMONSTRATES
================================================================================
1. FastAPI Route Definition:
- Define REST endpoints using FastAPI decorators
- Use query parameters, path parameters, and request bodies
- Return typed responses from handlers
2. Mediator Dependency Injection:
- Use FastAPI's Depends() to inject mediator into routes
- Create mediator factory function for dependency injection
- Mediator is created per request (or can be singleton)
3. Query Handler Integration:
- Map queries to handlers using queries_mapper
- Handlers return typed Response objects
- Responses are automatically serialized to JSON by FastAPI
4. Separation of Concerns:
- FastAPI handles HTTP layer (routing, serialization, validation)
- CQRS handles business logic (command/query processing)
- Clear separation enables independent testing and evolution
================================================================================
REQUIREMENTS
================================================================================
Make sure you have installed:
- fastapi
- uvicorn
- cqrs (this package)
- di (dependency injection)
================================================================================
"""
import logging
import typing
import di
import fastapi
import uvicorn
import cqrs
from cqrs.requests import bootstrap
logging.basicConfig(level=logging.DEBUG)
api_router = fastapi.APIRouter(prefix="/api")
class Hello(cqrs.Request):
msg: str
class World(cqrs.Response):
msg: str
class HelloWorldQueryHandler(cqrs.RequestHandler[Hello, World]):
@property
def events(self) -> typing.List[cqrs.Event]:
return []
async def handle(self, request: Hello) -> World:
return World(msg="Hello CQRS!")
def queries_mapper(mapper: cqrs.RequestMap) -> None:
"""Maps queries to handlers."""
mapper.bind(Hello, HelloWorldQueryHandler)
def mediator_factory() -> cqrs.RequestMediator:
return bootstrap.bootstrap(
di_container=di.Container(),
queries_mapper=queries_mapper,
)
@api_router.get("/hello", status_code=fastapi.status.HTTP_200_OK)
async def hello_world(
msg: typing.Text = fastapi.Query(default="Hello World"),
mediator: cqrs.RequestMediator = fastapi.Depends(mediator_factory),
) -> World:
return await mediator.send(Hello(msg=msg))
if __name__ == "__main__":
app = fastapi.FastAPI()
app.include_router(api_router)
uvicorn.run(app, host="0.0.0.0", port=8000)