# retoor <retoor@molodetz.nl>
import json
import logging
import os
from fastapi import HTTPException, Request
from devplacepy.database import get_int_setting
from devplacepy.services.base import BaseService, ConfigField
from devplacepy.services.openai_gateway import config
from devplacepy.services.openai_gateway.analytics import summary_metrics
from devplacepy.services.openai_gateway.gateway import GatewayRuntime
from devplacepy.utils import get_current_user
logger = logging.getLogger(__name__)
def _presented_key(request: Request) -> str:
key = request.headers.get("X-API-KEY")
if key:
return key.strip()
scheme, _, credentials = request.headers.get("Authorization", "").partition(" ")
if scheme.lower() == "bearer" and credentials.strip():
return credentials.strip()
return ""
class GatewayService(BaseService):
default_enabled = True
min_interval = 5
title = "OpenAI Gateway"
description = (
"An OpenAI-compatible LLM endpoint at /openai/v1/* that forwards requests to "
"the configured upstream (DeepSeek by default). Image content is described by "
"a vision model first so vision-less upstreams still work. Access is gated by "
"a static key or DevPlace credentials, and throughput scales with the "
"instances setting."
)
config_fields = [
ConfigField(
"gateway_upstream_url",
"Upstream URL",
type="url",
default=config.UPSTREAM_URL_DEFAULT,
help="OpenAI-compatible chat-completions endpoint requests are forwarded to.",
group="Upstream",
),
ConfigField(
"gateway_model",
"Model",
type="str",
default=config.MODEL_DEFAULT,
help="Model sent upstream.",
group="Upstream",
),
ConfigField(
"gateway_force_model",
"Force model",
type="bool",
default=True,
help="Override the client-requested model with the configured model.",
group="Upstream",
),
ConfigField(
"gateway_api_key",
"Upstream API key",
type="str",
default="",
help="The key currently in use; auto-migrated from DEEPSEEK_API_KEY or OPENROUTER_API_KEY on boot. Editable.",
group="Upstream",
),
ConfigField(
"gateway_timeout",
"Upstream timeout (seconds)",
type="int",
default=config.TIMEOUT_DEFAULT,
minimum=config.TIMEOUT_MIN,
help="Per-request upstream timeout. Minimum five minutes.",
group="Upstream",
),
ConfigField(
"gateway_instances",
"Instances (concurrency)",
type="int",
default=config.INSTANCES_DEFAULT,
minimum=1,
maximum=64,
help="Max concurrent upstream forwards per worker (connection pool + semaphore).",
group="Upstream",
),
ConfigField(
"gateway_vision_enabled",
"Vision augmentation",
type="bool",
default=True,
help="Describe image content via the vision model before forwarding.",
group="Vision",
),
ConfigField(
"gateway_vision_url",
"Vision URL",
type="url",
default=config.VISION_URL_DEFAULT,
help="OpenAI-compatible endpoint used to describe images.",
group="Vision",
),
ConfigField(
"gateway_vision_model",
"Vision model",
type="str",
default=config.VISION_MODEL_DEFAULT,
help="Vision-capable model name.",
group="Vision",
),
ConfigField(
"gateway_vision_key",
"Vision API key",
type="str",
default="",
help="The key currently in use; auto-migrated from OPENROUTER_API_KEY on boot. Editable.",
group="Vision",
),
ConfigField(
"gateway_vision_cache_size",
"Vision cache size",
type="int",
default=config.VISION_CACHE_SIZE_DEFAULT,
minimum=0,
help="Image-description LRU cache entries (0 disables caching).",
group="Vision",
),
ConfigField(
"gateway_require_auth",
"Require authentication",
type="bool",
default=True,
help="When off, the gateway is open to anyone.",
group="Access",
),
ConfigField(
"gateway_allow_admins",
"Allow admins",
type="bool",
default=True,
help="Admin users (API key / Bearer / Basic / session) may call the gateway.",
group="Access",
),
ConfigField(
"gateway_allow_users",
"Allow users",
type="bool",
default=True,
help="Any authenticated user may call the gateway with their own API key. "
"Devii operates a signed-in user's account with that user's key, so usage is "
"attributed and limitable per user.",
group="Access",
),
ConfigField(
"gateway_access_key",
"Static access key",
type="password",
default="",
secret=True,
help="A standalone key that always grants access (sent as X-API-KEY or Bearer).",
group="Access",
),
ConfigField(
"gateway_internal_key",
"Internal key",
type="str",
default="",
help="Auto-generated on boot. DevPlace's own services authenticate to the gateway "
"with this key. Clear it and restart to rotate.",
group="Access",
),
ConfigField(
"gateway_price_cache_hit_per_m",
"Chat price cache-hit / 1M ($)",
type="float",
default=config.PRICE_CACHE_HIT_PER_M_DEFAULT,
minimum=0,
help="Estimates chat cost when the upstream returns no native cost field (DeepSeek).",
group="Pricing",
),
ConfigField(
"gateway_price_cache_miss_per_m",
"Chat price cache-miss / 1M ($)",
type="float",
default=config.PRICE_CACHE_MISS_PER_M_DEFAULT,
minimum=0,
group="Pricing",
),
ConfigField(
"gateway_price_output_per_m",
"Chat price output / 1M ($)",
type="float",
default=config.PRICE_OUTPUT_PER_M_DEFAULT,
minimum=0,
group="Pricing",
),
ConfigField(
"gateway_vision_price_input_per_m",
"Vision price input / 1M ($)",
type="float",
default=config.VISION_PRICE_INPUT_PER_M_DEFAULT,
minimum=0,
help="Fallback only; used when the vision upstream returns no native cost.",
group="Pricing",
),
ConfigField(
"gateway_vision_price_output_per_m",
"Vision price output / 1M ($)",
type="float",
default=config.VISION_PRICE_OUTPUT_PER_M_DEFAULT,
minimum=0,
group="Pricing",
),
ConfigField(
"gateway_max_retries",
"Max retries",
type="int",
default=config.MAX_RETRIES_DEFAULT,
minimum=0,
maximum=10,
help="Retry attempts on timeout, connection error, or upstream 5xx.",
group="Reliability",
),
ConfigField(
"gateway_retry_backoff_ms",
"Retry backoff (ms)",
type="int",
default=config.RETRY_BACKOFF_MS_DEFAULT,
minimum=0,
help="Linear backoff multiplied by the attempt number.",
group="Reliability",
),
ConfigField(
"gateway_circuit_threshold",
"Circuit breaker threshold",
type="int",
default=config.CIRCUIT_THRESHOLD_DEFAULT,
minimum=0,
help="Consecutive upstream failures before the breaker opens (0 disables).",
group="Reliability",
),
ConfigField(
"gateway_circuit_cooldown_seconds",
"Circuit breaker cooldown (s)",
type="int",
default=config.CIRCUIT_COOLDOWN_SECONDS_DEFAULT,
minimum=1,
group="Reliability",
),
ConfigField(
"gateway_usage_retention_hours",
"Usage retention (hours)",
type="int",
default=config.USAGE_RETENTION_HOURS_DEFAULT,
minimum=1,
help="How long per-call usage rows are kept before pruning.",
group="Tracking",
),
ConfigField(
"gateway_model_context_map",
"Model context map (JSON)",
type="str",
default=json.dumps(config.MODEL_CONTEXT_MAP_DEFAULT),
help="JSON object mapping model name to max context tokens for utilization tracking.",
group="Tracking",
),
]
def __init__(self):
super().__init__(name="openai", interval_seconds=30)
self._runtime = None
def runtime(self) -> GatewayRuntime:
if self._runtime is None:
self._runtime = GatewayRuntime()
return self._runtime
def effective_config(self) -> dict:
cfg = self.get_config()
cfg["gateway_api_key"] = (
cfg["gateway_api_key"]
or os.environ.get("DEEPSEEK_API_KEY", "")
or os.environ.get("OPENROUTER_API_KEY", "")
)
cfg["gateway_vision_key"] = cfg["gateway_vision_key"] or os.environ.get(
"OPENROUTER_API_KEY", ""
)
return cfg
def authorize(self, request: Request) -> bool:
cfg = self.get_config()
if not cfg["gateway_require_auth"]:
return True
access_key = cfg["gateway_access_key"]
internal_key = cfg["gateway_internal_key"]
presented = _presented_key(request)
if presented and access_key and presented == access_key:
return True
if presented and internal_key and presented == internal_key:
return True
user = get_current_user(request)
if user:
if user.get("role") == "Admin" and cfg["gateway_allow_admins"]:
return True
if cfg["gateway_allow_users"]:
return True
return False
def resolve_owner(self, request: Request) -> tuple:
cfg = self.get_config()
presented = _presented_key(request)
if (
presented
and cfg["gateway_internal_key"]
and presented == cfg["gateway_internal_key"]
):
return ("internal", "devii")
if (
presented
and cfg["gateway_access_key"]
and presented == cfg["gateway_access_key"]
):
return ("key", "access")
user = get_current_user(request)
if user:
kind = "admin" if user.get("role") == "Admin" else "user"
return (kind, user.get("uid") or "unknown")
return ("anonymous", "anonymous")
async def handle(self, request: Request, subpath: str):
if not self.is_enabled():
raise HTTPException(status_code=503, detail="Gateway is disabled")
if not self.authorize(request):
self.log(f"Rejected {request.method} /{subpath}: unauthorized")
raise HTTPException(status_code=401, detail="Unauthorized")
cfg = self.effective_config()
runtime = self.runtime()
owner = self.resolve_owner(request)
user_agent = request.headers.get("user-agent", "")
if subpath == "chat/completions" and request.method == "POST":
try:
body = await request.json()
except Exception:
self.log("Rejected chat request: invalid JSON body")
raise HTTPException(status_code=400, detail="Invalid JSON body")
if not isinstance(body, dict):
self.log("Rejected chat request: JSON body was not an object")
raise HTTPException(status_code=400, detail="Invalid JSON body")
return await runtime.handle_chat(body, cfg, owner, user_agent, self.log)
body = await request.body()
content_type = request.headers.get("content-type", "")
return await runtime.handle_passthrough(
request.method,
subpath,
content_type,
body,
cfg,
owner,
user_agent,
self.log,
)
async def run_once(self) -> None:
if not self.is_enabled():
return
runtime = self.runtime()
runtime._ensure(self.effective_config())
runtime._ledger.sample_concurrency(runtime.in_flight)
retention = get_int_setting(
"gateway_usage_retention_hours", config.USAGE_RETENTION_HOURS_DEFAULT
)
ledger_removed, samples_removed = runtime._ledger.prune(retention)
if ledger_removed or samples_removed:
self.log(
f"Pruned {ledger_removed} usage rows and {samples_removed} concurrency samples"
)
async def on_disable(self) -> None:
if self._runtime is not None:
await self._runtime.aclose()
def collect_metrics(self) -> dict:
cfg = self.get_config()
m = (
self._runtime.metrics()
if self._runtime is not None
else {
"requests": 0,
"errors": 0,
"in_flight": 0,
"peak_in_flight": 0,
"vision_calls": 0,
"last_status": 0,
"last_latency_ms": 0,
"pool": 0,
"circuit_open": False,
}
)
s = summary_metrics()
stats = [
{"label": "Requests (lifetime)", "value": m["requests"]},
{"label": "In flight", "value": m["in_flight"]},
{"label": "Vision calls", "value": m["vision_calls"]},
{"label": "Last status", "value": m["last_status"] or "-"},
{"label": "Last latency", "value": f"{m['last_latency_ms']} ms"},
{"label": "Pool size", "value": m["pool"]},
{"label": "Circuit", "value": "open" if m["circuit_open"] else "closed"},
{"label": "Model", "value": cfg["gateway_model"]},
{"label": "Requests 24h", "value": s["requests"]},
{"label": "Success 24h", "value": f"{s['success_pct']}%"},
{"label": "Error rate 24h", "value": f"{s['error_pct']}%"},
{"label": "Cost this hour", "value": f"${s['cost_hour']:.4f}"},
{"label": "Cost 24h", "value": f"${s['cost_24h']:.2f}"},
{"label": "Tokens 24h", "value": s["tokens_24h"]},
{"label": "Avg latency 24h", "value": f"{s['avg_latency_ms']:.0f} ms"},
{"label": "Avg tokens/s 24h", "value": s["avg_tps"]},
{"label": "Peak concurrency 24h", "value": s["peak_concurrency"]},
{"label": "Top model 24h", "value": s["top_model"]},
{"label": "Top caller 24h", "value": s["top_caller"]},
]
return {"stats": stats}
Reference in New Issue View Git Blame Copy Permalink