First attempt to tink demo
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
This commit is contained in:
Henrik Jess Nielsen
415
src/routes/demo.py
Normal file
415
src/routes/demo.py
Normal file
@@ -0,0 +1,415 @@
|
||||
"""
|
||||
Demo routes — each /demo/step/N page shows one Tink API call
|
||||
with the live JSON response and a curl example.
|
||||
"""
|
||||
|
||||
import json
|
||||
import uuid
|
||||
import secrets
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import APIRouter, Request, Form, HTTPException
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from src.tink.client import TinkClient
|
||||
from src.config import get_settings
|
||||
from src import demo_data
|
||||
|
||||
router = APIRouter()
|
||||
templates = Jinja2Templates(directory="src/templates")
|
||||
|
||||
|
||||
def _client() -> TinkClient:
|
||||
s = get_settings()
|
||||
return TinkClient(
|
||||
client_id=s.tink_client_id,
|
||||
client_secret=s.tink_client_secret,
|
||||
redirect_uri=s.tink_redirect_uri,
|
||||
api_base=s.tink_api_base,
|
||||
link_base=s.tink_link_base,
|
||||
)
|
||||
|
||||
|
||||
def _session(request: Request) -> dict:
|
||||
return request.session.setdefault("demo", {})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Landing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/", response_class=HTMLResponse)
|
||||
async def index(request: Request):
|
||||
return templates.TemplateResponse("index.html", {"request": request})
|
||||
|
||||
|
||||
@router.get("/demo/reset")
|
||||
async def reset_demo(request: Request):
|
||||
"""Clear all demo session state and restart from Step 1."""
|
||||
request.session.pop("demo", None)
|
||||
return RedirectResponse("/demo/step/1")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 1 — Authenticate (client credentials)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/demo/step/1", response_class=HTMLResponse)
|
||||
async def step1(request: Request):
|
||||
client = _client()
|
||||
s = get_settings()
|
||||
error = None
|
||||
result = None
|
||||
curl_example = (
|
||||
f"curl -X POST https://api.tink.com/api/v1/oauth/token \\\n"
|
||||
f" -d 'client_id={s.tink_client_id}' \\\n"
|
||||
f" -d 'client_secret=***' \\\n"
|
||||
f" -d 'grant_type=client_credentials' \\\n"
|
||||
f" -d 'scope=user:create'"
|
||||
)
|
||||
try:
|
||||
result = await client.get_app_token(scope="user:create")
|
||||
_session(request)["app_token"] = result["access_token"]
|
||||
except Exception as e:
|
||||
error = str(e)
|
||||
|
||||
return templates.TemplateResponse("step.html", {
|
||||
"request": request,
|
||||
"step": 1,
|
||||
"title": "Authenticate",
|
||||
"subtitle": "Client Credentials Flow",
|
||||
"endpoint": "POST /api/v1/oauth/token",
|
||||
"api_version": "v1",
|
||||
"description": "Vi starter med at hente et app-level access token via Client Credentials flow. Dette token bruges til at oprette brugere.",
|
||||
"curl_example": curl_example,
|
||||
"result": result,
|
||||
"error": error,
|
||||
"next_step": 2,
|
||||
"prev_step": None,
|
||||
})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 2 — Create User
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/demo/step/2", response_class=HTMLResponse)
|
||||
async def step2_get(request: Request):
|
||||
sess = _session(request)
|
||||
s = get_settings()
|
||||
app_token = sess.get("app_token", "")
|
||||
external_user_id = f"moneycapp-demo-{secrets.token_hex(4)}"
|
||||
|
||||
body = json.dumps({"external_user_id": external_user_id, "market": "DK", "locale": "da_DK"}, indent=2)
|
||||
curl_example = (
|
||||
f"curl -X POST https://api.tink.com/api/v1/user/create \\\n"
|
||||
f" -H 'Authorization: Bearer $APP_TOKEN' \\\n"
|
||||
f" -H 'Content-Type: application/json' \\\n"
|
||||
f" -d '{body}'"
|
||||
)
|
||||
|
||||
error = None
|
||||
result = None
|
||||
if not app_token:
|
||||
error = "Mangler app token — gå tilbage til Step 1 og kør authentication først."
|
||||
else:
|
||||
try:
|
||||
client = _client()
|
||||
result = await client.create_user(app_token, external_user_id)
|
||||
sess["user_id"] = result.get("user_id", "")
|
||||
sess["external_user_id"] = external_user_id
|
||||
except Exception as e:
|
||||
error = str(e)
|
||||
|
||||
return templates.TemplateResponse("step.html", {
|
||||
"request": request,
|
||||
"step": 2,
|
||||
"title": "Opret Bruger",
|
||||
"subtitle": "Create Test Customer",
|
||||
"endpoint": "POST /api/v1/user/create",
|
||||
"api_version": "v1",
|
||||
"description": "Vi opretter en ny Tink-bruger med et unikt external_user_id — det er dit interne kunde-ID. Tink returnerer et user_id som vi bruger i de næste kald.",
|
||||
"curl_example": curl_example,
|
||||
"result": result,
|
||||
"error": error,
|
||||
"next_step": 3,
|
||||
"prev_step": 1,
|
||||
})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 3 — Connect Bank (Tink Link redirect)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
CONSOLE_CALLBACK = "https://console.tink.com/callback"
|
||||
|
||||
|
||||
@router.get("/demo/step/3", response_class=HTMLResponse)
|
||||
async def step3_get(request: Request):
|
||||
sess = _session(request)
|
||||
s = get_settings()
|
||||
client = _client()
|
||||
|
||||
tink_link_url = client.get_tink_link_url(market="DK")
|
||||
dev_tink_link_url = client.get_tink_link_url(
|
||||
market="DK", redirect_uri_override=CONSOLE_CALLBACK
|
||||
)
|
||||
error = None
|
||||
credentials = None
|
||||
cb_success = request.query_params.get("cb_success")
|
||||
cb_error = request.query_params.get("cb_error")
|
||||
|
||||
# Demo mode: auto-mark as connected with mock data
|
||||
if s.demo_mode and not sess.get("user_token"):
|
||||
sess["user_token"] = "demo-mode-token"
|
||||
sess["demo_mode"] = True
|
||||
|
||||
# Check if already connected (returning from callback)
|
||||
user_token = sess.get("user_token", "")
|
||||
if user_token:
|
||||
if sess.get("demo_mode"):
|
||||
credentials = demo_data.MOCK_CREDENTIALS
|
||||
cb_success = cb_success or "demo"
|
||||
else:
|
||||
try:
|
||||
credentials = await client.list_credentials(user_token)
|
||||
except Exception as e:
|
||||
error = str(e)
|
||||
|
||||
curl_example = (
|
||||
"# Simpelt Tink Link flow — ingen pre-oprettet bruger nødvendig\n"
|
||||
f"# Redirect brugeren direkte til:\n"
|
||||
f"https://link.tink.com/1.0/transactions/connect-accounts\n"
|
||||
f" ?client_id=$CLIENT_ID\n"
|
||||
f" &redirect_uri=$REDIRECT_URI\n"
|
||||
f" &market=DK\n"
|
||||
f" &scope=accounts:read,transactions:read,credentials:read\n\n"
|
||||
"# Tink Link redirecter tilbage med ?code=... som du exchangeer for user token:\n"
|
||||
"curl -X POST https://api.tink.com/api/v1/oauth/token \\\n"
|
||||
" -d 'grant_type=authorization_code' \\\n"
|
||||
" -d 'client_id=$CLIENT_ID' \\\n"
|
||||
" -d 'client_secret=$CLIENT_SECRET' \\\n"
|
||||
" -d 'code=$CODE'"
|
||||
)
|
||||
|
||||
return templates.TemplateResponse("step.html", {
|
||||
"request": request,
|
||||
"step": 3,
|
||||
"title": "Tilslut Bank",
|
||||
"subtitle": "Tink Link — Bank Connection",
|
||||
"endpoint": "Tink Link /1.0/transactions/connect-accounts",
|
||||
"api_version": "Link v1",
|
||||
"description": "Brugeren åbner Tink Link, vælger Tink Demo Bank og logger ind med test-credentials fra Console. Tink redirecter tilbage med en authorization_code som automatisk exchang'es til et user token.",
|
||||
"curl_example": curl_example,
|
||||
"result": credentials,
|
||||
"tink_link_url": tink_link_url,
|
||||
"dev_tink_link_url": dev_tink_link_url,
|
||||
"error": error or (f"Callback fejl: {cb_error}" if cb_error else None),
|
||||
"cb_success": cb_success,
|
||||
"next_step": 4,
|
||||
"prev_step": 2,
|
||||
})
|
||||
|
||||
|
||||
@router.post("/demo/step/3", response_class=HTMLResponse)
|
||||
async def step3_post(request: Request, code: str = Form(...)):
|
||||
"""Manual code entry — exchange a code obtained via console.tink.com/callback."""
|
||||
sess = _session(request)
|
||||
try:
|
||||
client = _client()
|
||||
tokens = await client.exchange_code_for_token(
|
||||
code=code.strip(),
|
||||
redirect_uri=CONSOLE_CALLBACK,
|
||||
)
|
||||
sess["user_token"] = tokens.get("access_token", "")
|
||||
return RedirectResponse("/demo/step/3?cb_success=1", status_code=303)
|
||||
except Exception as e:
|
||||
return RedirectResponse(f"/demo/step/3?cb_error={str(e)}", status_code=303)
|
||||
|
||||
|
||||
@router.get("/callback", response_class=HTMLResponse)
|
||||
async def tink_callback(request: Request, code: Optional[str] = None,
|
||||
error: Optional[str] = None,
|
||||
credentials_id: Optional[str] = None):
|
||||
"""Tink Link OAuth callback — exchange code for user token."""
|
||||
sess = _session(request)
|
||||
if error:
|
||||
return RedirectResponse(f"/demo/step/3?error={error}")
|
||||
if code:
|
||||
try:
|
||||
client = _client()
|
||||
tokens = await client.exchange_code_for_token(code)
|
||||
sess["user_token"] = tokens.get("access_token", "")
|
||||
except Exception as e:
|
||||
return RedirectResponse(f"/demo/step/3?cb_error={str(e)}")
|
||||
return RedirectResponse("/demo/step/3")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 4 — Accounts (v2)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/demo/step/4", response_class=HTMLResponse)
|
||||
async def step4(request: Request):
|
||||
sess = _session(request)
|
||||
s = get_settings()
|
||||
user_token = sess.get("user_token", "")
|
||||
error = None
|
||||
result = None
|
||||
is_demo = sess.get("demo_mode") or s.demo_mode
|
||||
|
||||
curl_example = (
|
||||
"curl https://api.tink.com/data/v2/accounts \\\n"
|
||||
" -H 'Authorization: Bearer $USER_TOKEN'"
|
||||
)
|
||||
|
||||
if not user_token and not is_demo:
|
||||
error = "Mangler user token — tilslut en bank i Step 3 først."
|
||||
elif is_demo:
|
||||
result = demo_data.MOCK_ACCOUNTS
|
||||
else:
|
||||
try:
|
||||
client = _client()
|
||||
result = await client.list_accounts(user_token)
|
||||
# Fall back to demo data if no accounts connected yet
|
||||
if not result.get("accounts"):
|
||||
result = demo_data.MOCK_ACCOUNTS
|
||||
sess["demo_mode"] = True
|
||||
except Exception as e:
|
||||
error = str(e)
|
||||
|
||||
return templates.TemplateResponse("step.html", {
|
||||
"request": request,
|
||||
"step": 4,
|
||||
"title": "Konti",
|
||||
"subtitle": "Account List with Balances",
|
||||
"endpoint": "GET /data/v2/accounts",
|
||||
"api_version": "v2 ✦",
|
||||
"description": "Henter brugerens konti via det nye v2 data endpoint. Returnerer account type, balance, currency og IBAN.",
|
||||
"curl_example": curl_example,
|
||||
"result": result,
|
||||
"error": error,
|
||||
"is_demo": is_demo or (result == demo_data.MOCK_ACCOUNTS),
|
||||
"next_step": 5,
|
||||
"prev_step": 3,
|
||||
})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 5 — Transactions (v2)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/demo/step/5", response_class=HTMLResponse)
|
||||
async def step5(request: Request, account_id: Optional[str] = None):
|
||||
sess = _session(request)
|
||||
s = get_settings()
|
||||
user_token = sess.get("user_token", "")
|
||||
is_demo = sess.get("demo_mode") or s.demo_mode
|
||||
error = None
|
||||
result = None
|
||||
|
||||
curl_example = (
|
||||
"curl 'https://api.tink.com/data/v2/transactions?pageSize=25' \\\n"
|
||||
" -H 'Authorization: Bearer $USER_TOKEN'"
|
||||
)
|
||||
|
||||
if not user_token and not is_demo:
|
||||
error = "Mangler user token — tilslut en bank i Step 3 først."
|
||||
elif is_demo:
|
||||
result = demo_data.MOCK_TRANSACTIONS
|
||||
else:
|
||||
try:
|
||||
client = _client()
|
||||
result = await client.list_transactions(user_token, account_id=account_id)
|
||||
if not result.get("transactions"):
|
||||
result = demo_data.MOCK_TRANSACTIONS
|
||||
sess["demo_mode"] = True
|
||||
except Exception as e:
|
||||
error = str(e)
|
||||
|
||||
return templates.TemplateResponse("step.html", {
|
||||
"request": request,
|
||||
"step": 5,
|
||||
"title": "Transaktioner",
|
||||
"subtitle": "Transaction History",
|
||||
"endpoint": "GET /data/v2/transactions",
|
||||
"api_version": "v2 ✦",
|
||||
"description": "Henter transaktioner via v2 endpoint med paginering. Returnerer amount, description, status (BOOKED/PENDING), og kategorisering.",
|
||||
"curl_example": curl_example,
|
||||
"result": result,
|
||||
"error": error,
|
||||
"is_demo": is_demo or (result == demo_data.MOCK_TRANSACTIONS),
|
||||
"next_step": 6,
|
||||
"prev_step": 4,
|
||||
})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 6 — Events (v2)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/demo/step/6", response_class=HTMLResponse)
|
||||
async def step6(request: Request):
|
||||
sess = _session(request)
|
||||
s = get_settings()
|
||||
user_token = sess.get("user_token", "")
|
||||
is_demo = sess.get("demo_mode") or s.demo_mode
|
||||
error = None
|
||||
result_booked = None
|
||||
result_all = None
|
||||
|
||||
curl_booked = (
|
||||
"curl 'https://api.tink.com/events/v2/account-booked-transactions?pageSize=10' \\\n"
|
||||
" -H 'Authorization: Bearer $USER_TOKEN'"
|
||||
)
|
||||
curl_all = (
|
||||
"curl 'https://api.tink.com/events/v2/account-transactions?pageSize=10' \\\n"
|
||||
" -H 'Authorization: Bearer $USER_TOKEN'"
|
||||
)
|
||||
|
||||
if not user_token and not is_demo:
|
||||
error = "Mangler user token — tilslut en bank i Step 3 først."
|
||||
elif is_demo:
|
||||
result_booked = demo_data.MOCK_EVENTS_BOOKED
|
||||
result_all = demo_data.MOCK_EVENTS_ALL
|
||||
else:
|
||||
try:
|
||||
client = _client()
|
||||
result_booked = await client.list_booked_transaction_events(user_token, page_size=10)
|
||||
except Exception as e:
|
||||
error = f"Booked events: {e}"
|
||||
try:
|
||||
client = _client()
|
||||
result_all = await client.list_account_transaction_events(user_token, page_size=10)
|
||||
except Exception as e:
|
||||
error = (error or "") + f" | All events: {e}"
|
||||
|
||||
return templates.TemplateResponse("step6.html", {
|
||||
"request": request,
|
||||
"step": 6,
|
||||
"title": "Events",
|
||||
"subtitle": "Real-time Event Feed",
|
||||
"error": error,
|
||||
"result_booked": result_booked,
|
||||
"result_all": result_all,
|
||||
"curl_booked": curl_booked,
|
||||
"curl_all": curl_all,
|
||||
"is_demo": is_demo,
|
||||
"app_base_url": s.app_base_url,
|
||||
"next_step": None,
|
||||
"prev_step": 5,
|
||||
})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Webhook receiver
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.post("/webhooks/tink")
|
||||
async def webhook_receiver(request: Request):
|
||||
"""Receive Tink webhook events (configure URL in Tink Console)."""
|
||||
body = await request.json()
|
||||
# In production you'd verify the signature and store events
|
||||
print(f"[WEBHOOK] {json.dumps(body, indent=2)}")
|
||||
return {"status": "received"}
|
||||
Reference in New Issue
Block a user