-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathrequest_response_types.py
More file actions
386 lines (295 loc) · 12.1 KB
/
request_response_types.py
File metadata and controls
386 lines (295 loc) · 12.1 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
"""
Example: Different Request and Response Types
This example demonstrates the flexibility of the CQRS library in supporting different
types of Request and Response implementations. The library supports both Pydantic-based
and Dataclass-based implementations, allowing you to choose the best fit for your needs.
Use case: Flexibility in choosing request/response implementations. You can use:
- PydanticRequest/PydanticResponse for validation and serialization features
- DCRequest/DCResponse for lightweight implementations without Pydantic dependency
- Mix and match different types based on your requirements
================================================================================
HOW TO RUN THIS EXAMPLE
================================================================================
Run the example:
python examples/request_response_types.py
The example will:
- Demonstrate Pydantic-based requests and responses
- Demonstrate Dataclass-based requests and responses
- Show mixed usage (Pydantic request with Dataclass response, etc.)
- Verify that all types work correctly with the mediator
================================================================================
WHAT THIS EXAMPLE DEMONSTRATES
================================================================================
1. PydanticRequest and PydanticResponse:
- Use Pydantic models for automatic validation
- Benefit from Pydantic's serialization features
- Type-safe with runtime validation
2. DCRequest and DCResponse:
- Use Python dataclasses for lightweight implementations
- No Pydantic dependency required
- Simple and straightforward
3. Mixed Usage:
- Combine Pydantic requests with Dataclass responses
- Combine Dataclass requests with Pydantic responses
- Flexibility to choose the best type for each use case
4. Type Compatibility:
- All request types implement IRequest interface
- All response types implement IResponse interface
- Mediator works seamlessly with all types
================================================================================
REQUIREMENTS
================================================================================
Make sure you have installed:
- cqrs (this package)
- di (dependency injection)
- pydantic (for PydanticRequest/PydanticResponse)
================================================================================
"""
import asyncio
import dataclasses
import logging
import typing
import di
import pydantic
import cqrs
from cqrs.requests import bootstrap
logging.basicConfig(level=logging.INFO)
# Storage for demonstration
USER_STORAGE: typing.Dict[str, typing.Dict[str, typing.Any]] = {}
PRODUCT_STORAGE: typing.Dict[str, typing.Dict[str, typing.Any]] = {}
ORDER_STORAGE: typing.Dict[str, typing.Dict[str, typing.Any]] = {}
# ============================================================================
# Pydantic-based Request and Response
# ============================================================================
class CreateUserCommand(cqrs.PydanticRequest):
"""Pydantic-based command with automatic validation."""
username: str
email: str
age: int = pydantic.Field(gt=0, le=120)
class UserResponse(cqrs.PydanticResponse):
"""Pydantic-based response with validation."""
user_id: str
username: str
email: str
age: int
class CreateUserCommandHandler(cqrs.RequestHandler[CreateUserCommand, UserResponse]):
"""Handler using Pydantic request and response."""
@property
def events(self) -> typing.Sequence[cqrs.IEvent]:
return []
async def handle(self, request: CreateUserCommand) -> UserResponse:
user_id = f"user_{len(USER_STORAGE) + 1}"
user_data = {
"user_id": user_id,
"username": request.username,
"email": request.email,
"age": request.age,
}
USER_STORAGE[user_id] = user_data
print(f"Created user with Pydantic: {user_data}")
return UserResponse(**user_data)
# ============================================================================
# Dataclass-based Request and Response
# ============================================================================
@dataclasses.dataclass
class CreateProductCommand(cqrs.DCRequest):
"""Dataclass-based command - lightweight, no Pydantic dependency."""
name: str
price: float
category: str
@dataclasses.dataclass
class ProductResponse(cqrs.DCResponse):
"""Dataclass-based response - simple and straightforward."""
product_id: str
name: str
price: float
category: str
class CreateProductCommandHandler(
cqrs.RequestHandler[CreateProductCommand, ProductResponse],
):
"""Handler using Dataclass request and response."""
@property
def events(self) -> typing.Sequence[cqrs.IEvent]:
return []
async def handle(self, request: CreateProductCommand) -> ProductResponse:
product_id = f"product_{len(PRODUCT_STORAGE) + 1}"
product_data = {
"product_id": product_id,
"name": request.name,
"price": request.price,
"category": request.category,
}
PRODUCT_STORAGE[product_id] = product_data
print(f"Created product with Dataclass: {product_data}")
return ProductResponse(**product_data)
# ============================================================================
# Mixed: Pydantic Request with Dataclass Response
# ============================================================================
class CreateOrderCommand(cqrs.PydanticRequest):
"""Pydantic request with validation."""
user_id: str
product_id: str
quantity: int = pydantic.Field(gt=0)
@dataclasses.dataclass
class OrderResponse(cqrs.DCResponse):
"""Dataclass response - lightweight."""
order_id: str
user_id: str
product_id: str
quantity: int
total_price: float
class CreateOrderCommandHandler(
cqrs.RequestHandler[CreateOrderCommand, OrderResponse],
):
"""Handler mixing Pydantic request with Dataclass response."""
@property
def events(self) -> typing.Sequence[cqrs.IEvent]:
return []
async def handle(self, request: CreateOrderCommand) -> OrderResponse:
if request.user_id not in USER_STORAGE:
raise ValueError(f"User {request.user_id} not found")
if request.product_id not in PRODUCT_STORAGE:
raise ValueError(f"Product {request.product_id} not found")
order_id = f"order_{len(ORDER_STORAGE) + 1}"
product = PRODUCT_STORAGE[request.product_id]
total_price = product["price"] * request.quantity
order_data = {
"order_id": order_id,
"user_id": request.user_id,
"product_id": request.product_id,
"quantity": request.quantity,
"total_price": total_price,
}
ORDER_STORAGE[order_id] = order_data
print(f"Created order (Pydantic request + Dataclass response): {order_data}")
return OrderResponse(**order_data)
# ============================================================================
# Mixed: Dataclass Request with Pydantic Response
# ============================================================================
@dataclasses.dataclass
class GetUserQuery(cqrs.DCRequest):
"""Dataclass query - simple and lightweight."""
user_id: str
class UserDetailsResponse(cqrs.PydanticResponse):
"""Pydantic response with validation."""
user_id: str
username: str
email: str
age: int
total_orders: int = 0
class GetUserQueryHandler(
cqrs.RequestHandler[GetUserQuery, UserDetailsResponse],
):
"""Handler mixing Dataclass request with Pydantic response."""
@property
def events(self) -> typing.Sequence[cqrs.IEvent]:
return []
async def handle(self, request: GetUserQuery) -> UserDetailsResponse:
if request.user_id not in USER_STORAGE:
raise ValueError(f"User {request.user_id} not found")
user = USER_STORAGE[request.user_id]
total_orders = sum(1 for order in ORDER_STORAGE.values() if order["user_id"] == request.user_id)
return UserDetailsResponse(
user_id=user["user_id"],
username=user["username"],
email=user["email"],
age=user["age"],
total_orders=total_orders,
)
# ============================================================================
# Mapping and Bootstrap
# ============================================================================
def commands_mapper(mapper: cqrs.RequestMap) -> None:
"""Register all command handlers."""
mapper.bind(CreateUserCommand, CreateUserCommandHandler)
mapper.bind(CreateProductCommand, CreateProductCommandHandler)
mapper.bind(CreateOrderCommand, CreateOrderCommandHandler)
def queries_mapper(mapper: cqrs.RequestMap) -> None:
"""Register all query handlers."""
mapper.bind(GetUserQuery, GetUserQueryHandler)
# ============================================================================
# Main Execution
# ============================================================================
async def main():
"""Demonstrate different request/response type combinations."""
mediator = bootstrap.bootstrap(
di_container=di.Container(),
commands_mapper=commands_mapper,
queries_mapper=queries_mapper,
)
print("=" * 80)
print("Demonstrating Different Request/Response Types")
print("=" * 80)
print()
# 1. Pydantic Request + Pydantic Response
print("1. Pydantic Request + Pydantic Response")
print("-" * 80)
user_response = await mediator.send(
CreateUserCommand(username="john_doe", email="john@example.com", age=30),
)
print(f"Response type: {type(user_response).__name__}")
print(f"Response data: {user_response.to_dict()}")
print()
# 2. Dataclass Request + Dataclass Response
print("2. Dataclass Request + Dataclass Response")
print("-" * 80)
product_response = await mediator.send(
CreateProductCommand(name="Laptop", price=999.99, category="Electronics"),
)
print(f"Response type: {type(product_response).__name__}")
print(f"Response data: {product_response.to_dict()}")
print()
# 3. Pydantic Request + Dataclass Response
print("3. Pydantic Request + Dataclass Response")
print("-" * 80)
order_response = await mediator.send(
CreateOrderCommand(
user_id=user_response.user_id,
product_id=product_response.product_id,
quantity=2,
),
)
print(f"Response type: {type(order_response).__name__}")
print(f"Response data: {order_response.to_dict()}")
print()
# 4. Dataclass Request + Pydantic Response
print("4. Dataclass Request + Pydantic Response")
print("-" * 80)
user_details = await mediator.send(GetUserQuery(user_id=user_response.user_id))
print(f"Response type: {type(user_details).__name__}")
print(f"Response data: {user_details.to_dict()}")
print()
# Demonstrate serialization/deserialization
print("=" * 80)
print("Serialization/Deserialization Demo")
print("=" * 80)
print()
# Serialize Pydantic response
user_dict = user_response.to_dict()
print(f"Pydantic response serialized: {user_dict}")
restored_user = UserResponse.from_dict(**user_dict)
print(f"Pydantic response restored: {restored_user}")
print()
# Serialize Dataclass response
product_dict = product_response.to_dict()
print(f"Dataclass response serialized: {product_dict}")
restored_product = ProductResponse.from_dict(**product_dict)
print(f"Dataclass response restored: {restored_product}")
print()
# Validation example with Pydantic
print("=" * 80)
print("Pydantic Validation Example")
print("=" * 80)
try:
# This should fail validation (age > 120)
await mediator.send(
CreateUserCommand(username="invalid", email="test@example.com", age=150),
)
except pydantic.ValidationError as e:
print(f"Validation error caught (expected): {e}")
print()
print("=" * 80)
print("All examples completed successfully!")
print("=" * 80)
if __name__ == "__main__":
asyncio.run(main())