Python Integration

1
# Your Turalogin API key from the dashboard
2
TURALOGIN_API_KEY=tl_live_xxxxxxxxxxxxx
3
 
4
# The URL where magic links will redirect to
5
# Development:
6
APP_LOGIN_VALIDATION_URL=http://localhost:8000/auth/callback
7
 
8
# Production (update for your domain):
9
# APP_LOGIN_VALIDATION_URL=https://myapp.com/auth/callback
10
 
11
# Your session secret
12
SESSION_SECRET=your-session-secret-here

1
from fastapi import APIRouter, HTTPException
2
from pydantic import BaseModel, EmailStr
3
import httpx
4
import os
5
 
6
router = APIRouter()
7
 
8
TURALOGIN_API_KEY = os.environ["TURALOGIN_API_KEY"]
9
TURALOGIN_API_URL = "https://api.turalogin.com/api/v1"
10
VALIDATION_URL = os.environ["APP_LOGIN_VALIDATION_URL"]
11
 
12
class StartAuthRequest(BaseModel):
13
    email: EmailStr
14
 
15
@router.post("/auth/start")
16
async def start_auth(request: StartAuthRequest):
17
    async with httpx.AsyncClient() as client:
18
        response = await client.post(
19
            f"{TURALOGIN_API_URL}/auth/start",
20
            headers={
21
                "Authorization": f"Bearer {TURALOGIN_API_KEY}",
22
                "Content-Type": "application/json",
23
            },
24
            json={
25
                "email": request.email,
26
                "validationUrl": VALIDATION_URL  # Where the magic link redirects to
27
            },
28
        )
29
        
30
        if response.status_code != 200:
31
            data = response.json()
32
            raise HTTPException(
33
                status_code=response.status_code,
34
                detail=data.get("error", "Failed to start authentication")
35
            )
36
        
37
        return {"success": True, "message": "Check your email for the login link"}

1
from fastapi import Response, Request
2
from fastapi.responses import RedirectResponse
3
import jwt
4
from datetime import datetime, timedelta
5
 
6
SESSION_SECRET = os.environ["SESSION_SECRET"]
7
 
8
@router.get("/auth/callback")
9
async def auth_callback(request: Request, response: Response, token: str = None):
10
    if not token:
11
        return RedirectResponse(url="/login?error=invalid_link")
12
    
13
    async with httpx.AsyncClient() as client:
14
        api_response = await client.post(
15
            f"{TURALOGIN_API_URL}/auth/verify",
16
            headers={
17
                "Authorization": f"Bearer {TURALOGIN_API_KEY}",
18
                "Content-Type": "application/json",
19
            },
20
            json={"sessionId": token},
21
        )
22
        
23
        if api_response.status_code != 200:
24
            return RedirectResponse(url="/login?error=verification_failed")
25
        
26
        data = api_response.json()
27
        user = data["user"]
28
        
29
        # Create your own session token
30
        session_token = jwt.encode(
31
            {
32
                "user_id": user["id"],
33
                "email": user["email"],
34
                "exp": datetime.utcnow() + timedelta(days=7),
35
            },
36
            SESSION_SECRET,
37
            algorithm="HS256",
38
        )
39
        
40
        # Create redirect response with session cookie
41
        redirect = RedirectResponse(url="/dashboard", status_code=302)
42
        redirect.set_cookie(
43
            key="session",
44
            value=session_token,
45
            httponly=True,
46
            secure=os.environ.get("ENV") == "production",
47
            samesite="lax",
48
            max_age=7 * 24 * 60 * 60,
49
        )
50
        
51
        return redirect

1
from flask import Flask, request, jsonify, session, redirect, make_response
2
import requests
3
import os
4
import jwt
5
from datetime import datetime, timedelta
6
 
7
app = Flask(__name__)
8
app.secret_key = os.environ["SESSION_SECRET"]
9
 
10
TURALOGIN_API_KEY = os.environ["TURALOGIN_API_KEY"]
11
TURALOGIN_API_URL = "https://api.turalogin.com/api/v1"
12
VALIDATION_URL = os.environ["APP_LOGIN_VALIDATION_URL"]
13
 
14
@app.route("/auth/start", methods=["POST"])
15
def start_auth():
16
    data = request.get_json()
17
    email = data.get("email")
18
    
19
    if not email:
20
        return jsonify({"error": "Email is required"}), 400
21
    
22
    response = requests.post(
23
        f"{TURALOGIN_API_URL}/auth/start",
24
        headers={
25
            "Authorization": f"Bearer {TURALOGIN_API_KEY}",
26
            "Content-Type": "application/json",
27
        },
28
        json={
29
            "email": email,
30
            "validationUrl": VALIDATION_URL
31
        },
32
    )
33
    
34
    if response.status_code != 200:
35
        return jsonify(response.json()), response.status_code
36
    
37
    return jsonify({"success": True, "message": "Check your email for the login link"})
38
 
39
@app.route("/auth/callback")
40
def auth_callback():
41
    token = request.args.get("token")
42
    
43
    if not token:
44
        return redirect("/login?error=invalid_link")
45
    
46
    response = requests.post(
47
        f"{TURALOGIN_API_URL}/auth/verify",
48
        headers={
49
            "Authorization": f"Bearer {TURALOGIN_API_KEY}",
50
            "Content-Type": "application/json",
51
        },
52
        json={"sessionId": token},
53
    )
54
    
55
    if response.status_code != 200:
56
        return redirect("/login?error=verification_failed")
57
    
58
    result = response.json()
59
    session["user_id"] = result["user"]["id"]
60
    session["email"] = result["user"]["email"]
61
    
62
    return redirect("/dashboard")

1
import httpx
2
from dataclasses import dataclass
3
import os
4
 
5
@dataclass
6
class TuraloginUser:
7
    id: str
8
    email: str
9
 
10
@dataclass
11
class AuthResult:
12
    token: str
13
    user: TuraloginUser
14
 
15
class TuraloginError(Exception):
16
    def __init__(self, message: str, code: str, status: int):
17
        super().__init__(message)
18
        self.code = code
19
        self.status = status
20
 
21
class TuraloginService:
22
    def __init__(self, api_key: str, validation_url: str):
23
        self.api_key = api_key
24
        self.validation_url = validation_url
25
        self.base_url = "https://api.turalogin.com/api/v1"
26
    
27
    async def _request(self, endpoint: str, data: dict) -> dict:
28
        async with httpx.AsyncClient() as client:
29
            response = await client.post(
30
                f"{self.base_url}{endpoint}",
31
                headers={
32
                    "Authorization": f"Bearer {self.api_key}",
33
                    "Content-Type": "application/json",
34
                },
35
                json=data,
36
            )
37
            
38
            result = response.json()
39
            
40
            if response.status_code != 200:
41
                raise TuraloginError(
42
                    message=result.get("error", "Unknown error"),
43
                    code=result.get("code", "UNKNOWN"),
44
                    status=response.status_code,
45
                )
46
            
47
            return result
48
    
49
    async def start_auth(self, email: str) -> dict:
50
        """Start authentication - sends magic link email."""
51
        return await self._request("/auth/start", {
52
            "email": email,
53
            "validationUrl": self.validation_url
54
        })
55
    
56
    async def verify_token(self, session_id: str) -> AuthResult:
57
        """Verify token from magic link and return auth result."""
58
        result = await self._request("/auth/verify", {"sessionId": session_id})
59
        return AuthResult(
60
            token=result["token"],
61
            user=TuraloginUser(**result["user"]),
62
        )
63
 
64
# Usage:
65
# turalogin = TuraloginService(
66
#     os.environ["TURALOGIN_API_KEY"],
67
#     os.environ["APP_LOGIN_VALIDATION_URL"]
68
# )
69
# await turalogin.start_auth("user@example.com")
70
# result = await turalogin.verify_token(token_from_callback)

1
from fastapi import Depends, HTTPException, Request
2
from typing import Optional
3
import jwt
4
import os
5
 
6
SESSION_SECRET = os.environ["SESSION_SECRET"]
7
 
8
class CurrentUser:
9
    def __init__(self, id: str, email: str):
10
        self.id = id
11
        self.email = email
12
 
13
async def get_current_user(request: Request) -> CurrentUser:
14
    """
15
    Dependency that extracts and validates the session cookie.
16
    Raises 401 if not authenticated.
17
    """
18
    session_token = request.cookies.get("session")
19
    
20
    if not session_token:
21
        raise HTTPException(
22
            status_code=401,
23
            detail="Authentication required"
24
        )
25
    
26
    try:
27
        payload = jwt.decode(
28
            session_token,
29
            SESSION_SECRET,
30
            algorithms=["HS256"]
31
        )
32
        return CurrentUser(
33
            id=payload["user_id"],
34
            email=payload["email"]
35
        )
36
    except jwt.ExpiredSignatureError:
37
        raise HTTPException(
38
            status_code=401,
39
            detail="Session expired"
40
        )
41
    except jwt.InvalidTokenError:
42
        raise HTTPException(
43
            status_code=401,
44
            detail="Invalid session"
45
        )
46
 
47
async def get_optional_user(request: Request) -> Optional[CurrentUser]:
48
    """
49
    Dependency that returns the current user or None.
50
    Does not raise an error if not authenticated.
51
    """
52
    try:
53
        return await get_current_user(request)
54
    except HTTPException:
55
        return None
56
 
57
# Usage in routes:
58
# @router.get("/me")
59
# async def get_me(user: CurrentUser = Depends(get_current_user)):
60
#     return {"id": user.id, "email": user.email}