wehub-resource-sync
184 lines
4.6 KiB
Markdown
184 lines
4.6 KiB
Markdown
# python-resource-management — detailed worked examples
|
|
|
|
## Advanced Patterns
|
|
|
|
### Pattern 5: Selective Exception Suppression
|
|
|
|
Only suppress specific, documented exceptions.
|
|
|
|
```python
|
|
class StreamWriter:
|
|
"""Writer that handles broken pipe gracefully."""
|
|
|
|
def __init__(self, stream) -> None:
|
|
self._stream = stream
|
|
|
|
def __enter__(self) -> "StreamWriter":
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc_val: BaseException | None,
|
|
exc_tb: TracebackType | None,
|
|
) -> bool:
|
|
"""Clean up, suppressing BrokenPipeError on shutdown."""
|
|
self._stream.close()
|
|
|
|
# Suppress BrokenPipeError (client disconnected)
|
|
# This is expected behavior, not an error
|
|
if exc_type is BrokenPipeError:
|
|
return True # Exception suppressed
|
|
|
|
return False # Propagate all other exceptions
|
|
```
|
|
|
|
### Pattern 6: Streaming with Accumulated State
|
|
|
|
Maintain both incremental chunks and accumulated state during streaming.
|
|
|
|
```python
|
|
from collections.abc import Generator
|
|
from dataclasses import dataclass, field
|
|
|
|
@dataclass
|
|
class StreamingResult:
|
|
"""Accumulated streaming result."""
|
|
|
|
chunks: list[str] = field(default_factory=list)
|
|
_finalized: bool = False
|
|
|
|
@property
|
|
def content(self) -> str:
|
|
"""Get accumulated content."""
|
|
return "".join(self.chunks)
|
|
|
|
def add_chunk(self, chunk: str) -> None:
|
|
"""Add chunk to accumulator."""
|
|
if self._finalized:
|
|
raise RuntimeError("Cannot add to finalized result")
|
|
self.chunks.append(chunk)
|
|
|
|
def finalize(self) -> str:
|
|
"""Mark stream complete and return content."""
|
|
self._finalized = True
|
|
return self.content
|
|
|
|
def stream_with_accumulation(
|
|
response: StreamingResponse,
|
|
) -> Generator[tuple[str, str], None, str]:
|
|
"""Stream response while accumulating content.
|
|
|
|
Yields:
|
|
Tuple of (accumulated_content, new_chunk) for each chunk.
|
|
|
|
Returns:
|
|
Final accumulated content.
|
|
"""
|
|
result = StreamingResult()
|
|
|
|
for chunk in response.iter_content():
|
|
result.add_chunk(chunk)
|
|
yield result.content, chunk
|
|
|
|
return result.finalize()
|
|
```
|
|
|
|
### Pattern 7: Efficient String Accumulation
|
|
|
|
Avoid O(n²) string concatenation when accumulating.
|
|
|
|
```python
|
|
def accumulate_stream(stream) -> str:
|
|
"""Efficiently accumulate stream content."""
|
|
# BAD: O(n²) due to string immutability
|
|
# content = ""
|
|
# for chunk in stream:
|
|
# content += chunk # Creates new string each time
|
|
|
|
# GOOD: O(n) with list and join
|
|
chunks: list[str] = []
|
|
for chunk in stream:
|
|
chunks.append(chunk)
|
|
return "".join(chunks) # Single allocation
|
|
```
|
|
|
|
### Pattern 8: Tracking Stream Metrics
|
|
|
|
Measure time-to-first-byte and total streaming time.
|
|
|
|
```python
|
|
import time
|
|
from collections.abc import Generator
|
|
|
|
def stream_with_metrics(
|
|
response: StreamingResponse,
|
|
) -> Generator[str, None, dict]:
|
|
"""Stream response while collecting metrics.
|
|
|
|
Yields:
|
|
Content chunks.
|
|
|
|
Returns:
|
|
Metrics dictionary.
|
|
"""
|
|
start = time.perf_counter()
|
|
first_chunk_time: float | None = None
|
|
chunk_count = 0
|
|
total_bytes = 0
|
|
|
|
for chunk in response.iter_content():
|
|
if first_chunk_time is None:
|
|
first_chunk_time = time.perf_counter() - start
|
|
|
|
chunk_count += 1
|
|
total_bytes += len(chunk.encode())
|
|
yield chunk
|
|
|
|
total_time = time.perf_counter() - start
|
|
|
|
return {
|
|
"time_to_first_byte_ms": round((first_chunk_time or 0) * 1000, 2),
|
|
"total_time_ms": round(total_time * 1000, 2),
|
|
"chunk_count": chunk_count,
|
|
"total_bytes": total_bytes,
|
|
}
|
|
```
|
|
|
|
### Pattern 9: Managing Multiple Resources with ExitStack
|
|
|
|
Handle a dynamic number of resources cleanly.
|
|
|
|
```python
|
|
from contextlib import ExitStack, AsyncExitStack
|
|
from pathlib import Path
|
|
|
|
def process_files(paths: list[Path]) -> list[str]:
|
|
"""Process multiple files with automatic cleanup."""
|
|
results = []
|
|
|
|
with ExitStack() as stack:
|
|
# Open all files - they'll all be closed when block exits
|
|
files = [stack.enter_context(open(p)) for p in paths]
|
|
|
|
for f in files:
|
|
results.append(f.read())
|
|
|
|
return results
|
|
|
|
async def process_connections(hosts: list[str]) -> list[dict]:
|
|
"""Process multiple async connections."""
|
|
results = []
|
|
|
|
async with AsyncExitStack() as stack:
|
|
connections = [
|
|
await stack.enter_async_context(connect_to_host(host))
|
|
for host in hosts
|
|
]
|
|
|
|
for conn in connections:
|
|
results.append(await conn.fetch_data())
|
|
|
|
return results
|
|
```
|