Cenário: usando o SDK de autenticação do Microsoft Entra ID (sidecar) no Python

Crie uma biblioteca de cliente em Python que se integre ao SDK de autenticação do Microsoft Entra ID (sidecar) para obter tokens e chamar APIs subsequentes. Em seguida, integre esse cliente aos aplicativos Flask, FastAPI ou Django para lidar com solicitações autenticadas.

Pré-requisitos

Uma conta Azure com uma assinatura ativa. Crie uma conta gratuitamente.
Python (versão 3.7 ou posterior) com pip instalado em seu computador de desenvolvimento.
SDK de autenticação do Microsoft Entra ID (sidecar) implantado e em execução no seu ambiente. Consulte o Guia de Instalação para obter instruções de instalação.
APIs downstream configuradas no SDK com URLs base e escopos necessários.
Permissões adequadas no Microsoft Entra ID - sua conta deve ter permissões para registrar aplicativos e conceder permissões de API.

Configuração

Antes de criar sua biblioteca de clientes, instale a dependência necessária para fazer solicitações HTTP:

pip install requests

Implementação da biblioteca de clientes

Crie uma classe de cliente reutilizável que encapsula chamadas HTTP para o SDK de Autenticação Microsoft Entra ID (sidecar). Essa classe manipula o encaminhamento de token, a configuração de solicitação e o tratamento de erros:

# 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')

Integração do Flask

Integre a biblioteca de clientes a um aplicativo Flask extraindo o token de entrada em uma função auxiliar e usando-o em manipuladores de rota para chamar APIs downstream:

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)

Integração do FastAPI

Para aplicativos FastAPI, use o sistema de injeção de dependência com a Header dependência para extrair e validar o token de autorização antes de passá-lo para manipuladores de rota:

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))

Integração do Django

Para aplicativos Django, crie classes de exibição que extraam o token de autorização de cabeçalhos de solicitação e use-o para chamar APIs downstream:

# 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)

Avançado: usando solicitações. Sessão

Para melhorar o desempenho e a resiliência, use um requests.Session objeto com lógica de repetição. Essa abordagem permite novas tentativas automáticas de falhas transitórias e pool de conexões para reduzir a sobrecarga:

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

Práticas recomendadas

Ao usar o SDK de Autenticação Microsoft Entra ID (sidecar) do Python, siga estas práticas para criar aplicativos confiáveis e manteneveis:

Reutilizar a Instância do Cliente: crie uma única SidecarClient instância e reutilize-a em todo o aplicativo, em vez de criar novas instâncias por solicitação. Isso melhora o desempenho e o uso de recursos.
Definir tempos limite apropriados: configure tempos limite de solicitação com base na latência da API downstream. Isso impede que seu aplicativo fique suspenso indefinidamente se o SDK ou o serviço downstream estiver lento.
Implemente o Tratamento de Erros: adicione a lógica de repetição e tratamento de erros adequada, especialmente para falhas transitórias. Distingue entre erros de cliente (4xx) e erros de servidor (5xx) para determinar as respostas apropriadas.
Use anotações de tipo: adicione anotações de tipo aos parâmetros das funções e valores de retorno para uma maior clareza de código e para detectar erros durante o desenvolvimento.
Habilitar o Pool de Conexões: use um requests.Session objeto para reutilização de conexão entre solicitações, o que reduz a sobrecarga e melhora a taxa de transferência para várias chamadas à API.

Comentários

Esta página foi útil?

Last updated on 2026-06-16