class AsyncOntoPlatform:
    """Async Ontology platform client"""

    def __init__(self, config: SDKConfig):
        self._config = config
        self._connection_manager = AsyncConnectionManager(config)
        self._interceptor_chain = InterceptorChain([
            AsyncAuthInterceptor(config.auth_provider),
            AsyncRetryInterceptor(config.max_retries),
            AsyncTracingInterceptor(),
            AsyncMetricsInterceptor(),
        ])

        self.objects = AsyncObjectClient(self._connection_manager, self._interceptor_chain)
        self.actions = AsyncActionClient(self._connection_manager, self._interceptor_chain)
        self.search = AsyncSearchClient(self._connection_manager, self._interceptor_chain)
        self.links = AsyncLinkClient(self._connection_manager, self._interceptor_chain)

    @classmethod
    async def connect(cls, url: str, **kwargs) -> "AsyncOntoPlatform":
        config = SDKConfig(platform_url=url, **kwargs)
        client = cls(config)
        await client._connection_manager.initialize()
        return client

    async def close(self) -> None:
        await self._connection_manager.close()

    async def __aenter__(self) -> "AsyncOntoPlatform":
        return self

    async def __aexit__(self, *args) -> None:
        await self.close()

class AsyncConnectionManager:
    def __init__(self, config: SDKConfig):
        self._config = config
        self._channel: grpc.aio.Channel | None = None
        self._semaphore = asyncio.Semaphore(config.max_connections)

    async def initialize(self) -> None:
        self._channel = grpc.aio.insecure_channel(
            self._config.platform_url,
            options=[
                ("grpc.max_receive_message_length", 50 * 1024 * 1024),
                ("grpc.keepalive_time_ms", 10000),
            ],
        )

    @asynccontextmanager
    async def connection(self):
        await self._semaphore.acquire()
        try:
            yield self._channel
        finally:
            self._semaphore.release()

class AsyncObjectClient:
    async def get_many(
        self, object_type: str, object_ids: list[str], concurrency: int = 10,
    ) -> list[OntologyObject]:
        semaphore = asyncio.Semaphore(concurrency)

        async def fetch_one(oid: str) -> OntologyObject:
            async with semaphore:
                return await self.get(object_type, oid)

        tasks = [fetch_one(oid) for oid in object_ids]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        objects = []
        errors = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                errors.append((object_ids[i], result))
            else:
                objects.append(result)

        if errors:
            logger.warning(f"Failed to fetch {len(errors)} objects")

        return objects

class AsyncBatchWriter:
    def __init__(self, client, batch_size=100, max_concurrent=5, flush_interval=1.0):
        self._client = client
        self._batch_size = batch_size
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._buffer: list[WriteOperation] = []
        self._flush_interval = flush_interval

    async def add(self, operation: WriteOperation) -> None:
        self._buffer.append(operation)
        if len(self._buffer) >= self._batch_size:
            await self.flush()

    async def flush(self) -> BatchResult:
        if not self._buffer:
            return BatchResult(total=0, succeeded=0, failed=0)
        batch = self._buffer[:self._batch_size]
        self._buffer = self._buffer[self._batch_size:]
        async with self._semaphore:
            return await self._client.batch_write(batch)

    async def __aenter__(self):
        self._flush_task = asyncio.create_task(self._auto_flush())
        return self

    async def __aexit__(self, *args):
        if self._flush_task:
            self._flush_task.cancel()
        while self._buffer:
            await self.flush()

class AsyncResultStream:
    def __init__(self, grpc_stream, buffer_size=100):
        self._stream = grpc_stream
        self._buffer = asyncio.Queue(maxsize=buffer_size)
        self._exhausted = False
        self._fetch_task = None

    def __aiter__(self):
        return self

    async def __anext__(self) -> OntologyObject:
        if self._fetch_task is None:
            self._fetch_task = asyncio.create_task(self._fetch_loop())
        try:
            item = await asyncio.wait_for(self._buffer.get(), timeout=30)
            if item is _SENTINEL:
                raise StopAsyncIteration
            return item
        except asyncio.TimeoutError:
            raise StopAsyncIteration

    async def _fetch_loop(self):
        try:
            async for response in self._stream:
                obj = self._deserialize(response)
                await self._buffer.put(obj)
        finally:
            await self._buffer.put(_SENTINEL)

class AsyncPipeline:
    def __init__(self):
        self._stages = []

    def map(self, func): ...
    def filter(self, predicate): ...
    def batch(self, size): ...

    async def execute(self, source: AsyncIterable) -> list:
        results = []
        async for item in self._apply_stages(source):
            results.append(item)
        return results

# Usage
results = await (
    AsyncPipeline()
    .filter(lambda emp: emp.salary > 50000)
    .map(lambda emp: {"name": emp.name, "salary": emp.salary})
    .batch(100)
    .execute(client.objects.Employee.stream())
)

class SyncOntoPlatform:
    """Synchronous wrapper for non-async scenarios"""

    @classmethod
    def connect(cls, url: str, **kwargs) -> "SyncOntoPlatform":
        loop = asyncio.new_event_loop()
        async_client = loop.run_until_complete(AsyncOntoPlatform.connect(url, **kwargs))
        wrapper = cls(async_client)
        wrapper._loop = loop
        return wrapper

    def _run(self, coro):
        return self._loop.run_until_complete(coro)

# Usage
client = SyncOntoPlatform.connect("https://platform.example.com", token="...")
employee = client.objects.get("Employee", "emp-001")
client.close()

class GracefulShutdown:
    async def shutdown(self, timeout: float = 30.0) -> None:
        self._client._accepting_requests = False

        if self._pending_tasks:
            done, pending = await asyncio.wait(self._pending_tasks, timeout=timeout)
            for task in pending:
                task.cancel()

        await self._client.close()

class TestAsyncSDK:
    @pytest.mark.asyncio
    async def test_concurrent_get_many(self):
        async with AsyncOntoPlatform.connect(test_url, token=token) as client:
            ids = [f"emp-{i:03d}" for i in range(100)]
            results = await client.objects.get_many("Employee", ids, concurrency=10)
            assert len(results) == 100

    @pytest.mark.asyncio
    async def test_batch_writer(self):
        async with AsyncOntoPlatform.connect(test_url, token=token) as client:
            async with AsyncBatchWriter(client.objects, batch_size=50) as writer:
                for i in range(200):
                    await writer.add(WriteOperation(
                        type="create", object_type="Employee",
                        properties={"name": f"Emp {i}", "department": "Test"},
                    ))

    @pytest.mark.asyncio
    async def test_streaming_backpressure(self):
        async with AsyncOntoPlatform.connect(test_url, token=token) as client:
            count = 0
            async for emp in client.objects.Employee.stream():
                count += 1
                if count >= 1000:
                    break
            assert count == 1000