Async Usage

Installation

python -m pip install "vedika-sdk==3.1.0"

Run the synchronous SDK in a worker

import asyncio, json
from vedika import VedikaClient
BIRTH = {"datetime": "1990-05-15T10:30:00+05:30", "latitude": 28.6139, "longitude": 77.2090, "timezone": "Asia/Kolkata"}
def fetch_chart(birth):
    client = VedikaClient(timeout=15, max_retries=0, cache_enabled=True)
    try:
        return client.get_birth_chart_v2("birth-chart", birth)
    finally:
        client.session.close()
async def main():
    envelope = await asyncio.to_thread(fetch_chart, BIRTH)
    print(json.dumps(envelope["data"], ensure_ascii=False))
if __name__ == "__main__":
    asyncio.run(main())

Concurrent Requests

# Reuse fetch_chart and BIRTH from above; each worker owns its session.
from concurrent.futures import ThreadPoolExecutor
async def fetch_many(workers, births):
    loop = asyncio.get_running_loop()
    return await asyncio.gather(*(loop.run_in_executor(workers, fetch_chart, b) for b in births))
with ThreadPoolExecutor(max_workers=2) as workers:
    results = asyncio.run(fetch_many(workers, [BIRTH, BIRTH]))
print([envelope["data"] for envelope in results])

Rate Limiting with Semaphore

# A concurrency cap is an application limit, not an account rate allowance.
async def fetch_with_rate_limit(births, max_concurrent=2):
    semaphore = asyncio.Semaphore(max_concurrent)
    async def one(birth):
        async with semaphore:
            return await asyncio.to_thread(fetch_chart, birth)
    return await asyncio.gather(*(one(b) for b in births))
# Valid dates only; select a small authorized batch before sending.
charts = asyncio.run(fetch_with_rate_limit([BIRTH, BIRTH]))

Async Streaming

def collect_stream():
    client = VedikaClient(timeout=15, max_retries=0)
    try:
        return list(client.ask_question_stream("Explain the chart's main themes.", BIRTH))
    finally:
        client.session.close()
async def stream_response():
    for payload in await asyncio.to_thread(collect_stream):
        print(payload)  # Raw SSE data strings, not typed event objects
# asyncio.run(stream_response())  # Sends a billable query when explicitly run

FastAPI Integration

# Reuse fetch_chart; install FastAPI separately in your server application.
from fastapi import FastAPI, HTTPException
from vedika import VedikaAPIError
app = FastAPI()
@app.post("/chart")
async def get_chart(birth: dict):
    try:
        return await asyncio.to_thread(fetch_chart, birth)
    except VedikaAPIError:
        raise HTTPException(502, "Calculation could not be completed.") from None
def fetch_match(male, female):
    client = VedikaClient(timeout=15, max_retries=0)
    try:
        return client.get_guna_milan(male, female)
    finally:
        client.session.close()
@app.post("/match")
async def match_charts(male: dict, female: dict):
    try:
        return await asyncio.to_thread(fetch_match, male, female)
    except VedikaAPIError:
        raise HTTPException(502, "Matching could not be completed.") from None

Async Error Handling

from vedika import AuthenticationError, RateLimitError, VedikaAPIError
async def guarded_fetch(birth):
    try:
        return await asyncio.to_thread(fetch_chart, birth)
    except (AuthenticationError, RateLimitError, VedikaAPIError):
        # Keep upstream details in protected diagnostics; do not retry blindly.
        print("Request failed; review the account configuration and limits.")
        raise

Validation, stored charts and cache isolation