0
# Unsafe Operations
1

2
Escape hatch utilities that allow breaking out of the functional programming world when necessary. These operations bypass the safety guarantees of containers and should be used judiciously.
3

4
## Capabilities
5

6
### IO Container Execution
7

8
Extract values from IO containers by executing their side effects.
9

10
```python { .api }
11
def unsafe_perform_io(container: IO[T]) -> T:
12
    """Execute IO container and extract value (unsafe!)"""
13
```
14

15
Usage examples:
16

17
```python
18
from returns.io import IO, impure
19
from returns.unsafe import unsafe_perform_io
20

21
# Create IO container
22
@impure
23
def read_config_file() -> dict:
24
    with open("config.json") as f:
25
        return {"database_url": "localhost", "debug": True}
26

27
io_config = read_config_file()  # IO[dict]
28

29
# Extract value unsafely  
30
config = unsafe_perform_io(io_config)  # dict: {"database_url": "localhost", "debug": True}
31

32
# Use extracted config
33
print(f"Connecting to: {config['database_url']}")
34
```
35

36
## When to Use Unsafe Operations
37

38
Unsafe operations should only be used in specific scenarios:
39

40
### 1. Application Boundaries
41

42
At the edges of your application where you need to interface with the outside world:
43

44
```python
45
from returns.io import IO, impure
46
from returns.unsafe import unsafe_perform_io
47
from returns.result import safe
48

49
@impure
50
def setup_database() -> str:
51
    # Side effect: establish database connection
52
    return "Database connected"
53

54
@safe
55
def process_user_data(data: dict) -> dict:
56
    # Pure business logic
57
    return {"processed": True, "user_id": data["id"]}
58

59
# Application entry point
60
def main():
61
    # Unsafe: execute side effects at application boundary
62
    db_status = unsafe_perform_io(setup_database())
63
    print(db_status)
64
    
65
    # Safe: pure business logic
66
    result = process_user_data({"id": 123})
67
    print(result)
68

69
if __name__ == "__main__":
70
    main()  # Only place where unsafe operations happen
71
```
72

73
### 2. Testing and Development
74

75
Extract values for testing purposes:
76

77
```python
78
from returns.io import IO, impure
79
from returns.unsafe import unsafe_perform_io
80

81
@impure  
82
def fetch_user_data(user_id: int) -> dict:
83
    # In real app: HTTP request
84
    return {"id": user_id, "name": "Test User"}
85

86
def test_user_processing():
87
    # Extract value for testing
88
    io_user = fetch_user_data(123)
89
    user_data = unsafe_perform_io(io_user)
90
    
91
    assert user_data["id"] == 123
92
    assert "name" in user_data
93
```
94

95
### 3. Legacy Code Integration
96

97
When integrating with existing non-functional codebases:
98

99
```python
100
from returns.io import IO, impure  
101
from returns.unsafe import unsafe_perform_io
102

103
# Legacy function expects regular values
104
def legacy_email_sender(config: dict, message: str) -> bool:
105
    # Existing imperative code
106
    print(f"Sending '{message}' with config: {config}")
107
    return True
108

109
@impure
110
def load_email_config() -> dict:
111
    return {"smtp_host": "smtp.example.com", "port": 587}
112

113
# Bridge between functional and imperative code
114
def send_notification(message: str) -> bool:
115
    config_io = load_email_config()
116
    config = unsafe_perform_io(config_io)  # Extract for legacy function
117
    return legacy_email_sender(config, message)
118
```
119

120
### 4. Performance Critical Sections
121

122
When container overhead is prohibitive (use sparingly):
123

124
```python
125
from returns.io import IO, impure
126
from returns.unsafe import unsafe_perform_io
127

128
@impure
129
def read_large_dataset() -> list[int]:
130
    # Expensive I/O operation
131
    return list(range(1000000))
132

133
def process_dataset_fast():
134
    # Sometimes you need raw performance
135
    dataset_io = read_large_dataset()
136
    dataset = unsafe_perform_io(dataset_io)
137
    
138
    # Time-critical processing on raw data
139
    return sum(x * x for x in dataset)
140
```
141

142
## Safety Guidelines
143

144
When using unsafe operations, follow these guidelines:
145

146
### 1. Minimize Usage
147

148
Keep unsafe operations at the absolute minimum:
149

150
```python
151
# BAD: Unsafe operations scattered throughout
152
def bad_example():
153
    config = unsafe_perform_io(load_config())
154
    users = unsafe_perform_io(fetch_users(config))
155
    for user in users:
156
        result = unsafe_perform_io(process_user(user))
157
        unsafe_perform_io(save_result(result))
158

159
# GOOD: Single unsafe operation at boundary  
160
def good_example():
161
    # Pure functional pipeline
162
    pipeline = (
163
        load_config()
164
        .bind(fetch_users)
165
        .bind(lambda users: sequence([process_user(u) for u in users]))
166
        .bind(save_all_results)
167
    )
168
    
169
    # Single unsafe execution
170
    unsafe_perform_io(pipeline)
171
```
172

173
### 2. Document Usage
174

175
Always document why unsafe operations are necessary:
176

177
```python
178
def critical_system_shutdown():
179
    """
180
    Performs emergency system shutdown.
181
    
182
    Uses unsafe_perform_io because:
183
    - Critical path where container overhead is prohibitive
184
    - Must execute immediately without functional composition
185
    - Called only in emergency scenarios
186
    """
187
    shutdown_io = initiate_shutdown()
188
    unsafe_perform_io(shutdown_io)  # Required for immediate execution
189
```
190

191
### 3. Isolate Side Effects
192

193
Keep unsafe operations isolated from pure code:
194

195
```python
196
# Pure functional core
197
def calculate_metrics(data: list[dict]) -> dict:
198
    return {"total": len(data), "average": sum(d["value"] for d in data) / len(data)}
199

200
# Unsafe shell around pure core
201
def metrics_service():
202
    data_io = fetch_data_from_api()
203
    data = unsafe_perform_io(data_io)  # Isolated unsafe operation
204
    
205
    metrics = calculate_metrics(data)  # Pure calculation
206
    
207
    save_metrics_io = save_metrics(metrics)
208
    unsafe_perform_io(save_metrics_io)  # Isolated unsafe operation
209
```
210

211
## Alternatives to Unsafe Operations
212

213
Before using unsafe operations, consider these alternatives:
214

215
1. **Compose with bind/map**: Keep operations in the container world
216
2. **Use IOResult**: Handle failures functionally
217
3. **Pipeline composition**: Chain operations without extraction
218
4. **Dependency injection**: Use context containers instead of extraction
219

220
Remember: unsafe operations break functional programming guarantees. Use them only when absolutely necessary and always at application boundaries.