Scénář: Použití sady SDK pro ověřování Microsoft Entra ID (sidecar) v Pythonu

Vytvořte klientskou knihovnu pro Python, která se integruje se sadou Microsoft Entra ID Auth SDK (sidecar) k získávání tokenů a volání navazujících rozhraní API. Potom tento klient integrujte do aplikací Flask, FastAPI nebo Django, které zpracovávají ověřené požadavky.

Požadavky

Účet Azure s aktivním předplatným. Vytvořte si bezplatný účet.
Python (verze 3.7 nebo novější) s pip nainstalovaným na vývojovém počítači.
Microsoft Entra ID Auth SDK (sidecar) nasazené a spuštěné ve vašem prostředí. Pokyny k instalaci najdete v průvodci instalací .
Podřízená rozhraní API nakonfigurovaná v sadě SDK se základními adresami URL a požadovanými obory
Příslušná oprávnění v Microsoft Entra ID – Váš účet musí mít oprávnění k registraci aplikací a udělení oprávnění k rozhraním API.

Nastavení

Před vytvořením klientské knihovny nainstalujte požadovanou závislost pro provádění požadavků HTTP:

pip install requests

Implementace klientské knihovny

Vytvořte opakovaně použitelnou klientskou třídu, která zabalí volání HTTP do sady Microsoft Entra ID Auth SDK (sidecar). Tato třída zpracovává předávání tokenů, konfiguraci požadavku a zpracování chyb:

# sidecar_client.py
import os
import json
import requests
from typing import Dict, Any, Optional, List
from urllib.parse import urlencode

class SidecarClient:
    """Client for interacting with the Microsoft Entra ID Auth SDK (sidecar)."""
    
    def __init__(self, base_url: Optional[str] = None, timeout: int = 10):
        self.base_url = base_url or os.getenv('SIDECAR_URL', 'http://localhost:5000')
        self.timeout = timeout
    
    def get_authorization_header(
        self,
        incoming_token: str,
        service_name: str,
        scopes: Optional[List[str]] = None,
        tenant: Optional[str] = None,
        agent_identity: Optional[str] = None,
        agent_username: Optional[str] = None
    ) -> str:
        """Get authorization header from the SDK."""
        params = {}
        
        if scopes:
            params['optionsOverride.Scopes'] = scopes
        
        if tenant:
            params['optionsOverride.AcquireTokenOptions.Tenant'] = tenant
        
        if agent_identity:
            params['AgentIdentity'] = agent_identity
            if agent_username:
                params['AgentUsername'] = agent_username
        
        response = requests.get(
            f"{self.base_url}/AuthorizationHeader/{service_name}",
            params=params,
            headers={'Authorization': incoming_token},
            timeout=self.timeout
        )
        
        response.raise_for_status()
        data = response.json()
        return data['authorizationHeader']
    
    def call_downstream_api(
        self,
        incoming_token: str,
        service_name: str,
        relative_path: str,
        method: str = 'GET',
        body: Optional[Dict[str, Any]] = None,
        scopes: Optional[List[str]] = None
    ) -> Any:
        """Call downstream API via the SDK."""
        params = {'optionsOverride.RelativePath': relative_path}
        
        if method != 'GET':
            params['optionsOverride.HttpMethod'] = method
        
        if scopes:
            params['optionsOverride.Scopes'] = scopes
        
        headers = {'Authorization': incoming_token}
        json_body = None
        
        if body:
            headers['Content-Type'] = 'application/json'
            json_body = body
        
        response = requests.request(
            method,
            f"{self.base_url}/DownstreamApi/{service_name}",
            params=params,
            headers=headers,
            json=json_body,
            timeout=self.timeout
        )
        
        response.raise_for_status()
        data = response.json()
        
        if data['statusCode'] >= 400:
            raise Exception(f"API error {data['statusCode']}: {data['content']}")
        
        return json.loads(data['content'])

# Usage
sidecar = SidecarClient(base_url='http://localhost:5000')

# Get authorization header
auth_header = sidecar.get_authorization_header(token, 'Graph')

# Call API
profile = sidecar.call_downstream_api(token, 'Graph', 'me')

Integrace Flasku

Integrujte klientskou knihovnu do aplikace Flask extrahováním příchozího tokenu do pomocné funkce a jeho použitím v obslužných rutinách tras pro volání podřízených rozhraní API:

from flask import Flask, request, jsonify
from sidecar_client import SidecarClient

app = Flask(__name__)
sidecar = SidecarClient()

def get_token():
    """Extract token from request."""
    token = request.headers.get('Authorization')
    if not token:
        raise ValueError('No authorization token provided')
    return token

@app.route('/api/profile')
def profile():
    try:
        token = get_token()
        profile_data = sidecar.call_downstream_api(
            token,
            'Graph',
            'me'
        )
        return jsonify(profile_data)
    except ValueError as e:
        return jsonify({'error': str(e)}), 401
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/messages')
def messages():
    try:
        token = get_token()
        messages_data = sidecar.call_downstream_api(
            token,
            'Graph',
            'me/messages?$top=10'
        )
        return jsonify(messages_data)
    except ValueError as e:
        return jsonify({'error': str(e)}), 401
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/messages/send', methods=['POST'])
def send_message():
    try:
        token = get_token()
        message = request.json
        
        result = sidecar.call_downstream_api(
            token,
            'Graph',
            'me/sendMail',
            method='POST',
            body={'message': message}
        )
        
        return jsonify({'success': True, 'result': result})
    except ValueError as e:
        return jsonify({'error': str(e)}), 401
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

Integrace FastAPI

Pro aplikace FastAPI použijte systém injektáže závislostí s Header k extrakci a ověření autorizačního tokenu před jeho předáním obslužným funkcím tras:

from fastapi import FastAPI, Header, HTTPException
from sidecar_client import SidecarClient
from typing import Optional

app = FastAPI()
sidecar = SidecarClient()

async def get_token(authorization: Optional[str] = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401, detail="No authorization token")
    return authorization

@app.get("/api/profile")
async def get_profile(token: str = Depends(get_token)):
    try:
        return sidecar.call_downstream_api(token, 'Graph', 'me')
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/messages")
async def get_messages(token: str = Depends(get_token)):
    try:
        return sidecar.call_downstream_api(
            token,
            'Graph',
            'me/messages?$top=10'
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Integrace Django

Pro aplikace Django vytvořte třídy zobrazení, které extrahují autorizační token z hlaviček požadavků a používají ho k volání podřízených rozhraní API:

# views.py
from django.http import JsonResponse
from django.views import View
from sidecar_client import SidecarClient

sidecar = SidecarClient()

class ProfileView(View):
    def get(self, request):
        token = request.META.get('HTTP_AUTHORIZATION')
        if not token:
            return JsonResponse({'error': 'No authorization token'}, status=401)
        
        try:
            profile = sidecar.call_downstream_api(token, 'Graph', 'me')
            return JsonResponse(profile)
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=500)

class MessagesView(View):
    def get(self, request):
        token = request.META.get('HTTP_AUTHORIZATION')
        if not token:
            return JsonResponse({'error': 'No authorization token'}, status=401)
        
        try:
            messages = sidecar.call_downstream_api(
                token,
                'Graph',
                'me/messages?$top=10'
            )
            return JsonResponse(messages)
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=500)

Pokročilé: použití požadavků. Sezení

Pokud chcete zvýšit výkon a odolnost, použijte objekt s logikou requests.Session opakování. Tento přístup umožňuje automatické opakování dočasných selhání a spojování připojení, čímž se snižuje režijní náročnost:

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

class SidecarClient:
    def __init__(self, base_url: Optional[str] = None):
        self.base_url = base_url or os.getenv('SIDECAR_URL', 'http://localhost:5000')
        
        # Configure session with retry logic
        self.session = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=0.3,
            status_forcelist=[500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount('http://', adapter)
        self.session.mount('https://', adapter)
    
    def call_downstream_api(self, token, service_name, relative_path, **kwargs):
        # Use self.session instead of requests
        response = self.session.get(...)
        return response

Osvědčené postupy

Při používání sady Microsoft Entra ID Auth SDK (sidecar) v Pythonu dodržujte tyto zásady, abyste vytvářeli spolehlivé a snadno udržovatelné aplikace:

Opakované použití instance klienta: Vytvořte jednu SidecarClient instanci a znovu ji použijte v celé aplikaci místo vytváření nových instancí na požadavek. Tím se zlepší výkon a využití prostředků.
Nastavení vhodných časových limitů: Nakonfigurujte časové limity požadavků na základě latence podřízené rozhraní API. To zabrání tomu, aby aplikace zůstala neomezeně zablokovaná, pokud je sada SDK nebo navazující služba pomalá.
Implementace zpracování chyb: Přidejte správné zpracování chyb a logiku opakování, zejména v případě přechodných selhání. Rozlišovat mezi chybami klienta (4xx) a chybami serveru (5xx) a určit odpovídající odpovědi.
Používejte typové návěští: Přidejte typová návěští k parametrům funkcí a návratovým hodnotám pro lepší srozumitelnost kódu a odhalování chyb během vývoje.
Povolit sdružování připojení: Použijte requests.Session objekt pro opakované použití připojení napříč požadavky, což snižuje režii a zlepšuje propustnost pro více volání rozhraní API.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-06-16