admin/OpenFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: saxo connector

Browse Source
This commit is contained in:
OpenSquared
2026-07-18 21:07:17 +02:00
parent d359480302
commit b986c8c525
4 changed files with 38 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
backend/services/saxo_auth.py
View File

@@ -1,10 +1,10 @@
"""
Saxo OpenAPI OAuth2 Authorization Code Grant — SIM environment only.
Saxo OpenAPI OAuth2 Authorization Code Grant — SIM or Live, set via SAXO_ENVIRONMENT.
Access tokens last ~20 min, refresh tokens ~40 min (SIM), so a background scheduler
(services/saxo_scheduler.py) must proactively refresh well before either expires —
this module only exposes the primitives (build the login URL, exchange code, refresh,
read a currently-valid token).
Access tokens last ~20 min, refresh tokens ~40 min (SIM figures — Live may differ), so a
background scheduler (services/saxo_scheduler.py) must proactively refresh well before
either expires — this module only exposes the primitives (build the login URL, exchange
code, refresh, read a currently-valid token).
"""
import os
import secrets
@@ -19,8 +19,17 @@ logger = logging.getLogger(__name__)
SAXO_APP_KEY = os.environ.get("SAXO_APP_KEY", "")
SAXO_APP_SECRET = os.environ.get("SAXO_APP_SECRET", "")
SAXO_REDIRECT_URI = os.environ.get("SAXO_REDIRECT_URI", "https://openfin.open-squared.tech/oauth/saxo/callback")
SAXO_AUTH_BASE_URL = os.environ.get("SAXO_AUTH_BASE_URL", "https://sim.logonvalidation.net")
SAXO_API_BASE_URL = os.environ.get("SAXO_API_BASE_URL", "https://gateway.saxobank.com/sim/openapi")
# SAXO_ENVIRONMENT=sim (default) or live — picks the matching auth/API host pair.
# Explicit SAXO_AUTH_BASE_URL/SAXO_API_BASE_URL env vars still win if set, for edge cases.
SAXO_ENVIRONMENT = os.environ.get("SAXO_ENVIRONMENT", "sim").strip().lower()
_DEFAULT_HOSTS = {
"sim": ("https://sim.logonvalidation.net", "https://gateway.saxobank.com/sim/openapi"),
"live": ("https://live.logonvalidation.net", "https://gateway.saxobank.com/openapi"),
}
_default_auth, _default_api = _DEFAULT_HOSTS.get(SAXO_ENVIRONMENT, _DEFAULT_HOSTS["sim"])
SAXO_AUTH_BASE_URL = os.environ.get("SAXO_AUTH_BASE_URL", _default_auth)
SAXO_API_BASE_URL = os.environ.get("SAXO_API_BASE_URL", _default_api)
# In-memory CSRF state for the OAuth login->callback round trip (single-user desktop app).
_pending_state: Optional[str] = None
@@ -64,7 +73,7 @@ def _store_token_response(token_resp: Dict[str, Any]):
refresh_token=token_resp["refresh_token"],
expires_at=expires_at,
refresh_expires_at=refresh_expires_at,
environment="sim",
environment=SAXO_ENVIRONMENT,
)
@@ -129,7 +138,7 @@ def get_status() -> Dict[str, Any]:
from services.database import get_saxo_tokens
stored = get_saxo_tokens()
if not stored or not stored.get("access_token"):
return {"connected": False, "configured": is_configured(), "environment": "sim"}
return {"connected": False, "configured": is_configured(), "environment": SAXO_ENVIRONMENT}
return {
"connected": True,
"configured": is_configured(),