0
# Operation Resolution
1
2
System for mapping OpenAPI operations to Python functions using various resolution strategies. Resolvers provide flexible ways to connect OpenAPI operation definitions to your Python implementation code.
3
4
## Capabilities
5
6
### Base Resolver
7
8
Basic operation resolver that maps operations to functions based on operationId.
9
10
```python { .api }
11
class Resolver:
12
def __init__(self, function_resolver=None):
13
"""
14
Initialize basic resolver.
15
16
Parameters:
17
- function_resolver: Custom function resolver callable
18
"""
19
20
def resolve(self, operation):
21
"""
22
Resolve an operation to a Python function.
23
24
Parameters:
25
- operation: Operation object to resolve
26
27
Returns:
28
Resolution object containing function and metadata
29
"""
30
31
def resolve_function_from_operation_id(self, operation_id: str):
32
"""
33
Resolve function directly from operation ID.
34
35
Parameters:
36
- operation_id: OpenAPI operation ID
37
38
Returns:
39
Python function or None if not found
40
"""
41
42
def resolve_operation_id(self, operation):
43
"""
44
Get operation ID from operation definition.
45
46
Parameters:
47
- operation: Operation object
48
49
Returns:
50
str: Operation ID
51
"""
52
```
53
54
### RestyResolver
55
56
RESTful operation resolver that maps operations to functions based on HTTP method and path patterns.
57
58
```python { .api }
59
class RestyResolver(Resolver):
60
def __init__(
61
self,
62
default_module_name: str,
63
collection_endpoint_name: str = "search"
64
):
65
"""
66
Initialize RESTful resolver.
67
68
Parameters:
69
- default_module_name: Default module to search for functions
70
- collection_endpoint_name: Name for collection endpoints
71
"""
72
73
def resolve_operation_id(self, operation):
74
"""
75
Generate operation ID based on HTTP method and path.
76
77
Uses RESTful conventions:
78
- GET /users -> get_users
79
- GET /users/{id} -> get_user
80
- POST /users -> post_users
81
- PUT /users/{id} -> put_user
82
- DELETE /users/{id} -> delete_user
83
84
Returns:
85
str: Generated operation ID
86
"""
87
```
88
89
### MethodResolver
90
91
Resolver for method-based operation handling, typically used with class-based views.
92
93
```python { .api }
94
class MethodResolver(Resolver):
95
def __init__(self, api_name: str, **kwargs):
96
"""
97
Initialize method resolver.
98
99
Parameters:
100
- api_name: Name of the API for method resolution
101
- **kwargs: Additional resolver options
102
"""
103
104
def resolve_operation_id_using_rest_semantics(self, operation):
105
"""
106
Resolve operation ID using REST semantic conventions.
107
108
Parameters:
109
- operation: Operation object
110
111
Returns:
112
str: Method-based operation ID
113
"""
114
```
115
116
### MethodViewResolver
117
118
Resolver for Flask MethodView-style class-based views.
119
120
```python { .api }
121
class MethodViewResolver(MethodResolver):
122
def __init__(self, api_name: str, **kwargs):
123
"""
124
Initialize MethodView resolver for Flask compatibility.
125
126
Parameters:
127
- api_name: API name for view resolution
128
- **kwargs: Additional resolver options
129
"""
130
```
131
132
### Resolution Result
133
134
Container for resolved operation information.
135
136
```python { .api }
137
class Resolution:
138
def __init__(self, function, operation_id: str):
139
"""
140
Initialize resolution result.
141
142
Parameters:
143
- function: Resolved Python function
144
- operation_id: Operation identifier
145
"""
146
147
@property
148
def function(self):
149
"""Resolved Python function"""
150
151
@property
152
def operation_id(self) -> str:
153
"""Operation identifier"""
154
```
155
156
### Custom Function Resolvers
157
158
Support for custom function resolution logic.
159
160
```python { .api }
161
def default_function_resolver(function_name: str):
162
"""
163
Default function resolver implementation.
164
165
Parameters:
166
- function_name: Name of function to resolve
167
168
Returns:
169
Function object or None if not found
170
"""
171
172
# Custom resolver example
173
def my_custom_resolver(function_name: str):
174
"""Custom function resolution logic"""
175
# Your custom logic here
176
pass
177
```
178
179
## Usage Examples
180
181
### Basic Resolver Usage
182
183
```python
184
from connexion import AsyncApp
185
from connexion.resolver import Resolver
186
187
# Use default resolver (looks for operationId in current module)
188
app = AsyncApp(__name__)
189
app.add_api('openapi.yaml') # Uses default Resolver
190
191
# Custom resolver with specific function resolver
192
def custom_function_resolver(function_name):
193
# Custom logic to find functions
194
return my_functions.get(function_name)
195
196
custom_resolver = Resolver(function_resolver=custom_function_resolver)
197
app.add_api('api.yaml', resolver=custom_resolver)
198
```
199
200
### RESTful Resolution
201
202
```python
203
from connexion.resolver import RestyResolver
204
205
# Create RESTful resolver
206
resolver = RestyResolver('api.handlers')
207
208
app = AsyncApp(__name__)
209
app.add_api('restful-api.yaml', resolver=resolver)
210
211
# With custom collection endpoint name
212
resolver = RestyResolver(
213
'api.handlers',
214
collection_endpoint_name='list'
215
)
216
app.add_api('api.yaml', resolver=resolver)
217
```
218
219
### Method-Based Resolution
220
221
```python
222
from connexion.resolver import MethodResolver
223
224
# For class-based views
225
resolver = MethodResolver('UserAPI')
226
227
app = AsyncApp(__name__)
228
app.add_api('users-api.yaml', resolver=resolver)
229
230
# Implementation class
231
class UserAPI:
232
def get(self, user_id=None):
233
if user_id:
234
return get_user(user_id)
235
return list_users()
236
237
def post(self):
238
return create_user(request.json)
239
240
def put(self, user_id):
241
return update_user(user_id, request.json)
242
243
def delete(self, user_id):
244
delete_user(user_id)
245
return NoContent, 204
246
```
247
248
### Custom Resolution Strategy
249
250
```python
251
from connexion.resolver import Resolver
252
253
class DatabaseResolver(Resolver):
254
"""Resolver that maps operations to database-stored procedures"""
255
256
def __init__(self, db_connection):
257
self.db = db_connection
258
super().__init__()
259
260
def resolve_function_from_operation_id(self, operation_id):
261
# Look up stored procedure by operation ID
262
procedure = self.db.get_procedure(operation_id)
263
if procedure:
264
return lambda **kwargs: procedure.execute(**kwargs)
265
return None
266
267
# Usage
268
db_resolver = DatabaseResolver(my_db_connection)
269
app.add_api('database-api.yaml', resolver=db_resolver)
270
```
271
272
### Module-Based Organization
273
274
```python
275
from connexion.resolver import RestyResolver
276
277
# Organize handlers in modules
278
# api/
279
# users.py - User operations
280
# products.py - Product operations
281
# orders.py - Order operations
282
283
resolver = RestyResolver('api')
284
285
# OpenAPI spec maps to functions like:
286
# GET /users -> api.users.get_users()
287
# POST /users -> api.users.post_users()
288
# GET /users/{id} -> api.users.get_user(id)
289
# GET /products -> api.products.get_products()
290
291
app.add_api('ecommerce-api.yaml', resolver=resolver)
292
```
293
294
### Resolver Error Handling
295
296
```python
297
from connexion.exceptions import ResolverError
298
299
def custom_resolver_error_handler(exception):
300
"""Handle resolver errors gracefully"""
301
if isinstance(exception, ResolverError):
302
return {
303
"error": "Operation not implemented",
304
"operation": exception.operation_id
305
}, 501
306
307
return {"error": "Internal server error"}, 500
308
309
app.add_api(
310
'api.yaml',
311
resolver=RestyResolver('handlers'),
312
resolver_error_handler=custom_resolver_error_handler
313
)
314
```
315
316
### Dynamic Function Loading
317
318
```python
319
import importlib
320
from connexion.resolver import Resolver
321
322
class DynamicResolver(Resolver):
323
"""Resolver that dynamically loads functions from modules"""
324
325
def __init__(self, module_prefix='api.v1'):
326
self.module_prefix = module_prefix
327
super().__init__()
328
329
def resolve_function_from_operation_id(self, operation_id):
330
# Split operation_id to get module and function
331
# e.g., "users_get_user" -> module="users", func="get_user"
332
parts = operation_id.split('_', 1)
333
if len(parts) != 2:
334
return None
335
336
module_name, func_name = parts
337
full_module = f"{self.module_prefix}.{module_name}"
338
339
try:
340
module = importlib.import_module(full_module)
341
return getattr(module, func_name, None)
342
except (ImportError, AttributeError):
343
return None
344
345
# Usage
346
dynamic_resolver = DynamicResolver('api.v2')
347
app.add_api('dynamic-api.yaml', resolver=dynamic_resolver)
348
```