0
# Context Operations
1
2
Context-dependent computations using the Reader monad pattern for dependency injection and environment passing. These containers enable clean dependency management without explicit parameter threading through function calls.
3
4
## Capabilities
5
6
### RequiresContext (Reader)
7
8
The basic Reader monad that represents a computation requiring some context/dependencies to produce a value.
9
10
```python { .api }
11
class RequiresContext[T, Deps]:
12
"""Reader monad for context-dependent computations"""
13
def __init__(self, func: Callable[[Deps], T]): ...
14
def __call__(self, deps: Deps) -> T: ...
15
def bind(self, func: Callable[[T], RequiresContext[U, Deps]]) -> RequiresContext[U, Deps]: ...
16
def map(self, func: Callable[[T], U]) -> RequiresContext[U, Deps]: ...
17
def apply(self, wrapped_func: RequiresContext[Callable[[T], U], Deps]) -> RequiresContext[U, Deps]: ...
18
19
@classmethod
20
def ask(cls) -> RequiresContext[Deps, Deps]:
21
"""Get the current context"""
22
23
@classmethod
24
def from_value(cls, value: T) -> RequiresContext[T, Deps]:
25
"""Lift a value into the context"""
26
27
# Type alias
28
Reader[T, Deps] = RequiresContext[T, Deps]
29
```
30
31
Usage examples:
32
33
```python
34
from returns.context import RequiresContext, Reader
35
36
# Define dependencies
37
class Config:
38
def __init__(self, db_url: str, api_key: str):
39
self.db_url = db_url
40
self.api_key = api_key
41
42
# Create context-dependent functions
43
def get_db_connection(config: Config) -> str:
44
return f"Connected to {config.db_url}"
45
46
def get_api_client(config: Config) -> str:
47
return f"API client with key {config.api_key}"
48
49
# Compose operations
50
def setup_services() -> Reader[tuple[str, str], Config]:
51
return (
52
RequiresContext(get_db_connection)
53
.bind(lambda db: RequiresContext(get_api_client).map(lambda api: (db, api)))
54
)
55
56
# Run with dependencies
57
config = Config("postgresql://localhost", "secret")
58
db_conn, api_client = setup_services()(config)
59
```
60
61
### RequiresContextResult (ReaderResult)
62
63
Combines Reader monad with Result for context-dependent computations that can fail.
64
65
```python { .api }
66
class RequiresContextResult[T, E, Deps]:
67
"""Reader + Result combination"""
68
def __init__(self, func: Callable[[Deps], Result[T, E]]): ...
69
def __call__(self, deps: Deps) -> Result[T, E]: ...
70
def bind(self, func: Callable[[T], RequiresContextResult[U, E, Deps]]) -> RequiresContextResult[U, E, Deps]: ...
71
def map(self, func: Callable[[T], U]) -> RequiresContextResult[U, E, Deps]: ...
72
def apply(self, wrapped_func: RequiresContextResult[Callable[[T], U], E, Deps]) -> RequiresContextResult[U, E, Deps]: ...
73
def alt(self, func: Callable[[E], RequiresContextResult[T, F, Deps]]) -> RequiresContextResult[T, F, Deps]: ...
74
def lash(self, func: Callable[[E], RequiresContextResult[T, F, Deps]]) -> RequiresContextResult[T, F, Deps]: ...
75
76
@classmethod
77
def ask(cls) -> RequiresContextResult[Deps, E, Deps]:
78
"""Get the current context"""
79
80
@classmethod
81
def from_success(cls, value: T) -> RequiresContextResult[T, E, Deps]:
82
"""Lift successful value into context"""
83
84
@classmethod
85
def from_failure(cls, error: E) -> RequiresContextResult[T, E, Deps]:
86
"""Lift failure into context"""
87
88
# Type aliases
89
ReaderResult[T, E, Deps] = RequiresContextResult[T, E, Deps]
90
ReaderResultE[T, Deps] = RequiresContextResult[T, Exception, Deps]
91
```
92
93
Usage examples:
94
95
```python
96
from returns.context import RequiresContextResult, ReaderResult
97
from returns.result import Success, Failure
98
99
# Context-dependent operations that can fail
100
def validate_config(config: Config) -> Result[Config, str]:
101
if not config.db_url:
102
return Failure("Database URL required")
103
if not config.api_key:
104
return Failure("API key required")
105
return Success(config)
106
107
def connect_database() -> ReaderResult[str, str, Config]:
108
return RequiresContextResult(
109
lambda config: validate_config(config).map(
110
lambda cfg: f"Connected to {cfg.db_url}"
111
)
112
)
113
114
# Chain operations
115
def setup_with_validation() -> ReaderResult[str, str, Config]:
116
return connect_database().bind(
117
lambda conn: RequiresContextResult.from_success(f"Setup complete: {conn}")
118
)
119
120
# Run with context
121
result = setup_with_validation()(config) # Success("Setup complete: Connected to...")
122
```
123
124
### RequiresContextIOResult (ReaderIOResult)
125
126
Combines Reader, IO, and Result for context-dependent side effects that can fail.
127
128
```python { .api }
129
class RequiresContextIOResult[T, E, Deps]:
130
"""Reader + IO + Result combination"""
131
def __init__(self, func: Callable[[Deps], IOResult[T, E]]): ...
132
def __call__(self, deps: Deps) -> IOResult[T, E]: ...
133
def bind(self, func: Callable[[T], RequiresContextIOResult[U, E, Deps]]) -> RequiresContextIOResult[U, E, Deps]: ...
134
def map(self, func: Callable[[T], U]) -> RequiresContextIOResult[U, E, Deps]: ...
135
def apply(self, wrapped_func: RequiresContextIOResult[Callable[[T], U], E, Deps]) -> RequiresContextIOResult[U, E, Deps]: ...
136
def alt(self, func: Callable[[E], RequiresContextIOResult[T, F, Deps]]) -> RequiresContextIOResult[T, F, Deps]: ...
137
138
@classmethod
139
def ask(cls) -> RequiresContextIOResult[Deps, E, Deps]:
140
"""Get the current context"""
141
142
# Type aliases
143
ReaderIOResult[T, E, Deps] = RequiresContextIOResult[T, E, Deps]
144
ReaderIOResultE[T, Deps] = RequiresContextIOResult[T, Exception, Deps]
145
```
146
147
### RequiresContextFutureResult (ReaderFutureResult)
148
149
Combines Reader with Future and Result for async context-dependent operations.
150
151
```python { .api }
152
class RequiresContextFutureResult[T, E, Deps]:
153
"""Reader + Future + Result combination"""
154
def __init__(self, func: Callable[[Deps], FutureResult[T, E]]): ...
155
def __call__(self, deps: Deps) -> FutureResult[T, E]: ...
156
async def bind(self, func: Callable[[T], RequiresContextFutureResult[U, E, Deps]]) -> RequiresContextFutureResult[U, E, Deps]: ...
157
async def map(self, func: Callable[[T], U]) -> RequiresContextFutureResult[U, E, Deps]: ...
158
async def apply(self, wrapped_func: RequiresContextFutureResult[Callable[[T], U], E, Deps]) -> RequiresContextFutureResult[U, E, Deps]: ...
159
async def alt(self, func: Callable[[E], RequiresContextFutureResult[T, F, Deps]]) -> RequiresContextFutureResult[T, F, Deps]: ...
160
161
@classmethod
162
def ask(cls) -> RequiresContextFutureResult[Deps, E, Deps]:
163
"""Get the current context"""
164
165
# Type aliases
166
ReaderFutureResult[T, E, Deps] = RequiresContextFutureResult[T, E, Deps]
167
ReaderFutureResultE[T, Deps] = RequiresContextFutureResult[T, Exception, Deps]
168
```
169
170
## Context Management Utilities
171
172
```python { .api }
173
# Type alias for no dependencies
174
NoDeps = Any
175
176
# Pointfree operations for context
177
def bind_context(func: Callable[[T], RequiresContext[U, Deps]]) -> Callable[[RequiresContext[T, Deps]], RequiresContext[U, Deps]]: ...
178
179
def modify_env(func: Callable[[Deps], NewDeps]) -> Callable[[RequiresContext[T, Deps]], RequiresContext[T, NewDeps]]: ...
180
```
181
182
## Usage Patterns
183
184
### Dependency Injection
185
186
```python
187
from returns.context import RequiresContext
188
189
class Database:
190
def get_user(self, user_id: int) -> dict:
191
return {"id": user_id, "name": "John"}
192
193
class Logger:
194
def log(self, message: str) -> None:
195
print(f"LOG: {message}")
196
197
class Dependencies:
198
def __init__(self, db: Database, logger: Logger):
199
self.db = db
200
self.logger = logger
201
202
def get_user_name(user_id: int) -> RequiresContext[str, Dependencies]:
203
def impl(deps: Dependencies) -> str:
204
deps.logger.log(f"Fetching user {user_id}")
205
user = deps.db.get_user(user_id)
206
return user["name"]
207
return RequiresContext(impl)
208
209
# Usage
210
deps = Dependencies(Database(), Logger())
211
name = get_user_name(123)(deps) # "John"
212
```
213
214
### Configuration Management
215
216
```python
217
from returns.context import RequiresContextResult
218
from returns.result import Success, Failure
219
220
class AppConfig:
221
def __init__(self, environment: str, debug: bool):
222
self.environment = environment
223
self.debug = debug
224
225
def get_feature_flag(flag_name: str) -> RequiresContextResult[bool, str, AppConfig]:
226
def impl(config: AppConfig) -> Result[bool, str]:
227
if config.environment == "production":
228
return Success(False) # Disable in production
229
elif config.environment == "development":
230
return Success(True) # Enable in development
231
else:
232
return Failure(f"Unknown environment: {config.environment}")
233
return RequiresContextResult(impl)
234
235
# Chain operations
236
def should_use_new_feature() -> RequiresContextResult[bool, str, AppConfig]:
237
return get_feature_flag("new_ui").bind(
238
lambda enabled: RequiresContextResult.from_success(
239
enabled and True # Additional logic
240
)
241
)
242
```
243
244
### Error Handling with Context
245
246
```python
247
from returns.context import RequiresContextResult
248
from returns.pointfree import bind_context
249
250
def safe_operation(x: int) -> RequiresContextResult[int, str, AppConfig]:
251
def impl(config: AppConfig) -> Result[int, str]:
252
if config.debug and x < 0:
253
return Failure("Debug mode: negative values not allowed")
254
return Success(x * 2)
255
return RequiresContextResult(impl)
256
257
# Compose with pointfree style
258
pipeline = bind_context(safe_operation)
259
result = pipeline(RequiresContextResult.from_success(5))
260
```
261
262
Context operations enable clean separation of business logic from infrastructure concerns, making code more testable and maintainable by explicitly managing dependencies through the type system.