5 เทคนิค Python จัดการทรัพยากร (Resource Orchestration) อย่างมีประสิทธิภาพ

async def build_dashboards_for_batch(user_ids: list[str], enabled_backends: list[str]) -> list[dict]:
    dashboards: list[dict] = []
 
async with asyncio.TaskGroup() as tg:
        async def run_one(uid: str) -> None:
            dashboard = await build_dashboard(uid, enabled_backends)
            dashboards.append(dashboard)
 
for uid in user_ids:
            tg.create_task(run_one(uid))
 
return dashboards

_semaphores: dict[str, asyncio.Semaphore] = {
    name: asyncio.Semaphore(cfg["capacity"]) for name, cfg in BACKEND_CONFIG.items()
}
 
@asynccontextmanager
async def acquire_connection(backend_name: str):
    semaphore = _semaphores[backend_name]
    async with semaphore:
        conn = await BackendConnection(backend_name).open()
        try:
            yield conn
        finally:
            await conn.close()

async with AsyncExitStack() as stack:
    connections = {
        name: await stack.enter_async_context(acquire_connection(name))
        for name in enabled_backends
    }
    # ... ใช้ `connections` ตามจำนวนที่มีอยู่จริง

try:
    async with asyncio.timeout(overall_timeout):
        async with asyncio.TaskGroup() as tg:
            async def run_one(name: str, conn) -> None:
                try:
                    async with asyncio.timeout(per_backend_timeout):
                        results[name] = await conn.query(user_id)
                except (TimeoutError, ConnectionError) as e:
                    errors[name] = str(e)
 
for name, conn in connections.items():
                tg.create_task(run_one(name, conn))
except TimeoutError:
    errors["_overall"] = f"dashboard build exceeded {overall_timeout}s overall budget"

# backends.py
import asyncio
import random
from dataclasses import dataclass
 
random.seed(11)
 
@dataclass
class BackendStats:
    open_connections: int = 0
    max_concurrent_open: int = 0
    max_concurrent_in_flight: int = 0
    in_flight: int = 0
    total_calls: int = 0
    total_failures: int = 0
 
STATS: dict[str, BackendStats] = {}
 
BACKEND_CONFIG = {
    "pricing_api":  {"latency": (0.02, 0.05), "capacity": 20, "failure_rate": 0.0},
    "positions_db": {"latency": (0.05, 0.10), "capacity": 10, "failure_rate": 0.0},
    "news_feed":    {"latency": (0.15, 0.25), "capacity": 5,  "failure_rate": 0.0},
    "risk_model":   {"latency": (0.30, 0.50), "capacity": 3,  "failure_rate": 0.15},
}
 
for name in BACKEND_CONFIG:
    STATS[name] = BackendStats()
 
class BackendConnection:
    def __init__(self, backend_name: str):
        self.backend_name = backend_name
        self._config = BACKEND_CONFIG[backend_name]
 
async def open(self) -> "BackendConnection":
        await asyncio.sleep(0.01)
        stats = STATS[self.backend_name]
        stats.open_connections += 1
        stats.max_concurrent_open = max(stats.max_concurrent_open, stats.open_connections)
        return self
 
async def close(self) -> None:
        await asyncio.sleep(0.005)
        STATS[self.backend_name].open_connections -= 1
 
async def query(self, request_id: str) -> dict:
        stats = STATS[self.backend_name]
        stats.in_flight += 1
        stats.max_concurrent_in_flight = max(stats.max_concurrent_in_flight, stats.in_flight)
        stats.total_calls += 1
        try:
            low, high = self._config["latency"]
            await asyncio.sleep(random.uniform(low, high))
            if random.random() < self._config["failure_rate"]:
                stats.total_failures += 1
                raise ConnectionError(f"{self.backend_name} timed out for request {request_id}")
            return {"backend": self.backend_name, "request_id": request_id, "data": f"result-from-{self.backend_name}"}
        finally:
            stats.in_flight -= 1
 
# pool.py
import asyncio
from contextlib import asynccontextmanager
from backends import BackendConnection, BACKEND_CONFIG
 
_semaphores: dict[str, asyncio.Semaphore] = {
    name: asyncio.Semaphore(cfg["capacity"]) for name, cfg in BACKEND_CONFIG.items()
}
 
@asynccontextmanager
async def acquire_connection(backend_name: str):
    semaphore = _semaphores[backend_name]
    async with semaphore:
        conn = await BackendConnection(backend_name).open()
        try:
            yield conn
        finally:
            await conn.close()
 
# orchestrator.py
import asyncio
from contextlib import AsyncExitStack
from pool import acquire_connection
 
async def build_dashboard(user_id: str, enabled_backends: list[str],
                           per_backend_timeout: float = 0.6,
                           overall_timeout: float = 1.0) -> dict:
    results: dict[str, dict] = {}
    errors: dict[str, str] = {}
 
async with AsyncExitStack() as stack:
        connections = {
            name: await stack.enter_async_context(acquire_connection(name))
            for name in enabled_backends
        }
 
try:
            async with asyncio.timeout(overall_timeout):
                async with asyncio.TaskGroup() as tg:
                    async def run_one(name: str, conn) -> None:
                        try:
                            async with asyncio.timeout(per_backend_timeout):
                                results[name] = await conn.query(user_id)
                        except (TimeoutError, ConnectionError) as e:
                            errors[name] = str(e)
 
for name, conn in connections.items():
                        tg.create_task(run_one(name, conn))
        except TimeoutError:
            errors["_overall"] = f"dashboard build exceeded {overall_timeout}s overall budget"
 
return {"user_id": user_id, "results": results, "errors": errors}
 
async def build_dashboards_for_batch(user_ids: list[str], enabled_backends: list[str]) -> list[dict]:
    dashboards: list[dict] = []
 
async with asyncio.TaskGroup() as tg:
        async def run_one(uid: str) -> None:
            dashboard = await build_dashboard(uid, enabled_backends)
            dashboards.append(dashboard)
 
for uid in user_ids:
            tg.create_task(run_one(uid))
 
return dashboards

python3 -c "
import asyncio
from orchestrator import build_dashboards_for_batch
from backends import BACKEND_CONFIG
 
async def main():
    user_ids = [f'user_{i}' for i in range(30)]
    dashboards = await build_dashboards_for_batch(user_ids, list(BACKEND_CONFIG.keys()))
    print(f'Completed {len(dashboards)} dashboards')
    errors = sum(len(d['errors']) for d in dashboards)
    print(f'Total backend errors: {errors} (risk_model has a 15% simulated failure rate)')
 
asyncio.run(main())
"