SSE Parsing 과 Dual Adapter

import httpx, json
from typing import AsyncIterator

async def stream_gemini_api_key(
    prompt: str, api_key: str, model: str = 'gemini-2.5-flash',
) -> AsyncIterator[str]:
    url = (
        f'https://generativelanguage.googleapis.com/v1beta/models/{model}'
        ':streamGenerateContent?alt=sse'
    )
    body = {'contents': [{'role': 'user', 'parts': [{'text': prompt}]}]}

    async with httpx.AsyncClient(timeout=120) as client:
        async with client.stream('POST', url,
            headers={'x-goog-api-key': api_key, 'Content-Type': 'application/json'},
            json=body,
        ) as resp:
            async for line in resp.aiter_lines():
                if not line.startswith('data: '):
                    continue
                chunk = json.loads(line[6:])
                for cand in chunk.get('candidates', []):
                    for part in cand.get('content', {}).get('parts', []):
                        if text := part.get('text'):
                            yield text

async def stream_gemini_oauth(
    prompt: str, access_token: str, project: str,
    model: str = 'gemini-2.5-flash',
) -> AsyncIterator[str]:
    url = 'https://cloudcode-pa.googleapis.com/v1internal:streamGenerateContent?alt=sse'
    body = {
        'model':   model,
        'project': project,
        'request': {
            'contents': [{'role': 'user', 'parts': [{'text': prompt}]}],
            'generationConfig': {},
        },
    }

    async with httpx.AsyncClient(timeout=120) as client:
        async with client.stream('POST', url,
            headers={
                'Authorization': f'Bearer {access_token}',
                'Content-Type':  'application/json',
            },
            json=body,
        ) as resp:
            async for line in resp.aiter_lines():
                if not line.startswith('data: '):
                    continue
                chunk = json.loads(line[6:])
                # OAuth wraps the response inside chunk['response']
                resp_obj = chunk.get('response', chunk)
                for cand in resp_obj.get('candidates', []):
                    for part in cand.get('content', {}).get('parts', []):
                        if text := part.get('text'):
                            yield text

class GeminiDualAdapter:
    def __init__(self, api_key: str, oauth_creds_path: str):
        self.api_key = api_key
        self.oauth_creds_path = oauth_creds_path
        self.fallback_active = False  # session-sticky

    async def stream(self, prompt: str):
        if not self.fallback_active:
            try:
                token   = get_access_token()  # from previous lesson
                project = load_code_assist(token)
                async for text in stream_gemini_oauth(
                    prompt, token, project,
                ):
                    yield text
                return
            except Exception as e:
                # Toast it visibly — never silent
                print(f'[OAuth failed: {e}. Falling back to API key, sticky for session.]')
                self.fallback_active = True

        # API key path
        async for text in stream_gemini_api_key(prompt, self.api_key):
            yield text