mirror of
https://github.com/wassname/openrouter-python-sdk-retry-errors.git
synced 2026-07-30 12:20:57 +08:00
Initial commit: OpenRouter Python SDK
Auto-generated Python SDK for OpenRouter API, providing comprehensive client library with: - Chat completions and streaming support - Embeddings API - Model and provider information - API key management - Analytics and usage tracking - OAuth authentication flows - Full type hints and error handling
This commit is contained in:
@@ -0,0 +1,136 @@
|
||||
"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT."""
|
||||
|
||||
from typing import (
|
||||
Any,
|
||||
Dict,
|
||||
List,
|
||||
Optional,
|
||||
)
|
||||
from httpx import Headers
|
||||
from pydantic import BaseModel
|
||||
from pydantic.fields import FieldInfo
|
||||
|
||||
from .metadata import (
|
||||
HeaderMetadata,
|
||||
find_field_metadata,
|
||||
)
|
||||
|
||||
from .values import _is_set, _populate_from_globals, _val_to_string
|
||||
|
||||
|
||||
def get_headers(headers_params: Any, gbls: Optional[Any] = None) -> Dict[str, str]:
|
||||
headers: Dict[str, str] = {}
|
||||
|
||||
globals_already_populated = []
|
||||
if _is_set(headers_params):
|
||||
globals_already_populated = _populate_headers(headers_params, gbls, headers, [])
|
||||
if _is_set(gbls):
|
||||
_populate_headers(gbls, None, headers, globals_already_populated)
|
||||
|
||||
return headers
|
||||
|
||||
|
||||
def _populate_headers(
|
||||
headers_params: Any,
|
||||
gbls: Any,
|
||||
header_values: Dict[str, str],
|
||||
skip_fields: List[str],
|
||||
) -> List[str]:
|
||||
globals_already_populated: List[str] = []
|
||||
|
||||
if not isinstance(headers_params, BaseModel):
|
||||
return globals_already_populated
|
||||
|
||||
param_fields: Dict[str, FieldInfo] = headers_params.__class__.model_fields
|
||||
for name in param_fields:
|
||||
if name in skip_fields:
|
||||
continue
|
||||
|
||||
field = param_fields[name]
|
||||
f_name = field.alias if field.alias is not None else name
|
||||
|
||||
metadata = find_field_metadata(field, HeaderMetadata)
|
||||
if metadata is None:
|
||||
continue
|
||||
|
||||
value, global_found = _populate_from_globals(
|
||||
name, getattr(headers_params, name), HeaderMetadata, gbls
|
||||
)
|
||||
if global_found:
|
||||
globals_already_populated.append(name)
|
||||
value = _serialize_header(metadata.explode, value)
|
||||
|
||||
if value != "":
|
||||
header_values[f_name] = value
|
||||
|
||||
return globals_already_populated
|
||||
|
||||
|
||||
def _serialize_header(explode: bool, obj: Any) -> str:
|
||||
if not _is_set(obj):
|
||||
return ""
|
||||
|
||||
if isinstance(obj, BaseModel):
|
||||
items = []
|
||||
obj_fields: Dict[str, FieldInfo] = obj.__class__.model_fields
|
||||
for name in obj_fields:
|
||||
obj_field = obj_fields[name]
|
||||
obj_param_metadata = find_field_metadata(obj_field, HeaderMetadata)
|
||||
|
||||
if not obj_param_metadata:
|
||||
continue
|
||||
|
||||
f_name = obj_field.alias if obj_field.alias is not None else name
|
||||
|
||||
val = getattr(obj, name)
|
||||
if not _is_set(val):
|
||||
continue
|
||||
|
||||
if explode:
|
||||
items.append(f"{f_name}={_val_to_string(val)}")
|
||||
else:
|
||||
items.append(f_name)
|
||||
items.append(_val_to_string(val))
|
||||
|
||||
if len(items) > 0:
|
||||
return ",".join(items)
|
||||
elif isinstance(obj, Dict):
|
||||
items = []
|
||||
|
||||
for key, value in obj.items():
|
||||
if not _is_set(value):
|
||||
continue
|
||||
|
||||
if explode:
|
||||
items.append(f"{key}={_val_to_string(value)}")
|
||||
else:
|
||||
items.append(key)
|
||||
items.append(_val_to_string(value))
|
||||
|
||||
if len(items) > 0:
|
||||
return ",".join([str(item) for item in items])
|
||||
elif isinstance(obj, List):
|
||||
items = []
|
||||
|
||||
for value in obj:
|
||||
if not _is_set(value):
|
||||
continue
|
||||
|
||||
items.append(_val_to_string(value))
|
||||
|
||||
if len(items) > 0:
|
||||
return ",".join(items)
|
||||
elif _is_set(obj):
|
||||
return f"{_val_to_string(obj)}"
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
def get_response_headers(headers: Headers) -> Dict[str, List[str]]:
|
||||
res: Dict[str, List[str]] = {}
|
||||
for k, v in headers.items():
|
||||
if not k in res:
|
||||
res[k] = []
|
||||
|
||||
res[k].append(v)
|
||||
return res
|
||||
Reference in New Issue
Block a user