CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-apache-airflow-providers-fab

Flask App Builder (FAB) authentication and authorization provider for Apache Airflow

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

api-endpoints.mddocs/

API Endpoints

REST API endpoints for programmatic user and role management through HTTP requests. These endpoints provide a web API interface for managing users, roles, and permissions in the FAB authentication system.

Capabilities

User Endpoints

REST API operations for user account management.

def get_users(
    limit: int = 100,
    offset: int = 0,
    order_by: str = "id"
) -> dict:
    """
    Get a list of users.
    
    Args:
        limit: Maximum number of users to return.
        offset: Number of users to skip.
        order_by: Field to order results by.
        
    Returns:
        Dictionary containing user list and metadata.
    """

def get_user(user_id: int) -> dict:
    """
    Get a specific user by ID.
    
    Args:
        user_id: The user's ID.
        
    Returns:
        Dictionary containing user details.
    """

def patch_user(user_id: int, data: dict) -> dict:
    """
    Update a user's information.
    
    Args:
        user_id: The user's ID.
        data: Dictionary containing fields to update.
        
    Returns:
        Dictionary containing updated user details.
    """

def delete_user(user_id: int) -> dict:
    """
    Delete a user account.
    
    Args:
        user_id: The user's ID.
        
    Returns:
        Dictionary containing operation status.
    """

def post_user(data: dict) -> dict:
    """
    Create a new user account.
    
    Args:
        data: Dictionary containing user details.
        
    Returns:
        Dictionary containing created user details.
    """

Role and Permission Endpoints

REST API operations for role and permission management.

def get_roles(
    limit: int = 100,
    offset: int = 0,
    order_by: str = "id"
) -> dict:
    """
    Get a list of roles.
    
    Args:
        limit: Maximum number of roles to return.
        offset: Number of roles to skip.
        order_by: Field to order results by.
        
    Returns:
        Dictionary containing role list and metadata.
    """

def get_role(role_id: int) -> dict:
    """
    Get a specific role by ID.
    
    Args:
        role_id: The role's ID.
        
    Returns:
        Dictionary containing role details and permissions.
    """

def patch_role(role_id: int, data: dict) -> dict:
    """
    Update a role's information.
    
    Args:
        role_id: The role's ID.  
        data: Dictionary containing fields to update.
        
    Returns:
        Dictionary containing updated role details.
    """

def delete_role(role_id: int) -> dict:
    """
    Delete a role.
    
    Args:
        role_id: The role's ID.
        
    Returns:
        Dictionary containing operation status.
    """

def post_role(data: dict) -> dict:
    """
    Create a new role.
    
    Args:
        data: Dictionary containing role details.
        
    Returns:
        Dictionary containing created role details.
    """

def get_permissions(
    limit: int = 100,
    offset: int = 0
) -> dict:
    """
    Get a list of available permissions.
    
    Args:
        limit: Maximum number of permissions to return.
        offset: Number of permissions to skip.
        
    Returns:
        Dictionary containing permission list and metadata.
    """

Usage Examples

User Management via API

import requests
import json

# API base URL
base_url = "http://localhost:8080/auth/fab/v1"

# Headers for JSON requests
headers = {
    "Content-Type": "application/json",
    "Authorization": "Basic <base64-encoded-credentials>"
}

# Get all users
response = requests.get(f"{base_url}/users", headers=headers)
users = response.json()
print(f"Total users: {users['total_entries']}")

# Get specific user
user_id = 1
response = requests.get(f"{base_url}/users/{user_id}", headers=headers)
user = response.json()
print(f"User: {user['username']} ({user['email']})")

# Create new user
new_user = {
    "username": "newuser",
    "first_name": "New",
    "last_name": "User", 
    "email": "newuser@example.com",
    "active": True,
    "roles": [{"name": "Viewer"}]
}
response = requests.post(f"{base_url}/users", headers=headers, json=new_user)
created_user = response.json()
print(f"Created user ID: {created_user['id']}")

# Update user
update_data = {
    "first_name": "Updated",
    "active": False
}
response = requests.patch(f"{base_url}/users/{user_id}", headers=headers, json=update_data)
updated_user = response.json()
print(f"Updated user: {updated_user['first_name']}")

# Delete user  
response = requests.delete(f"{base_url}/users/{user_id}", headers=headers)
print(f"Delete status: {response.status_code}")

Role Management via API

# Get all roles
response = requests.get(f"{base_url}/roles", headers=headers)
roles = response.json()
print(f"Available roles: {[role['name'] for role in roles['roles']]}")

# Get specific role with permissions
role_id = 1
response = requests.get(f"{base_url}/roles/{role_id}", headers=headers)
role = response.json()
print(f"Role: {role['name']}")
print(f"Permissions: {len(role['permissions'])}")

# Create new role
new_role = {
    "name": "DataAnalyst"
}
response = requests.post(f"{base_url}/roles", headers=headers, json=new_role)
created_role = response.json()
print(f"Created role: {created_role['name']}")

# Update role permissions
role_update = {
    "permissions": [
        {"action": {"name": "can_read"}, "resource": {"name": "DAG"}},
        {"action": {"name": "can_read"}, "resource": {"name": "DagRun"}}
    ]
}
response = requests.patch(f"{base_url}/roles/{role_id}", headers=headers, json=role_update)
updated_role = response.json()
print(f"Updated role permissions: {len(updated_role['permissions'])}")

Permission Discovery

# Get all available permissions
response = requests.get(f"{base_url}/permissions", headers=headers)
permissions = response.json()

print("Available permissions:")
for perm in permissions['permissions']:
    action = perm['action']['name']
    resource = perm['resource']['name'] 
    print(f"  {action} on {resource}")

# Filter by resource type
dag_permissions = [
    perm for perm in permissions['permissions'] 
    if perm['resource']['name'] == 'DAG'
]
print(f"DAG permissions: {len(dag_permissions)}")

Error Handling

def handle_api_response(response):
    """Handle API response with proper error checking."""
    if response.status_code == 200:
        return response.json()
    elif response.status_code == 401:
        raise Exception("Authentication failed")
    elif response.status_code == 403:
        raise Exception("Insufficient permissions")
    elif response.status_code == 404:
        raise Exception("Resource not found")
    elif response.status_code == 422:
        error_data = response.json()
        raise Exception(f"Validation error: {error_data['message']}")
    else:
        raise Exception(f"API error: {response.status_code}")

# Usage with error handling
try:
    response = requests.get(f"{base_url}/users/999", headers=headers)
    user = handle_api_response(response)
    print(f"User: {user['username']}")
except Exception as e:
    print(f"Error: {e}")

Pagination and Filtering

# Paginate through users
page_size = 10
offset = 0
all_users = []

while True:
    params = {
        "limit": page_size,
        "offset": offset,
        "order_by": "username"
    }
    response = requests.get(f"{base_url}/users", headers=headers, params=params)
    data = response.json()
    
    all_users.extend(data['users'])
    
    if len(data['users']) < page_size:
        break
    offset += page_size

print(f"Retrieved {len(all_users)} users total")

Bulk Operations

# Create multiple users
users_to_create = [
    {
        "username": f"user{i}",
        "first_name": f"User{i}",
        "last_name": "Test",
        "email": f"user{i}@example.com",
        "active": True,
        "roles": [{"name": "Viewer"}]
    }
    for i in range(1, 6)
]

created_users = []
for user_data in users_to_create:
    response = requests.post(f"{base_url}/users", headers=headers, json=user_data)
    if response.status_code == 200:
        created_users.append(response.json())
    else:
        print(f"Failed to create user {user_data['username']}: {response.status_code}")

print(f"Successfully created {len(created_users)} users")

API Response Formats

User Response Format

{
    "id": 1,
    "username": "admin",
    "first_name": "Admin",
    "last_name": "User", 
    "email": "admin@example.com",
    "active": true,
    "login_count": 42,
    "fail_login_count": 0,
    "created_on": "2023-01-01T00:00:00Z",
    "changed_on": "2023-01-15T12:30:00Z",
    "roles": [
        {
            "id": 1,
            "name": "Admin"
        }
    ]
}

Role Response Format

{
    "id": 1,
    "name": "Admin",
    "permissions": [
        {
            "id": 1,
            "action": {
                "id": 1,
                "name": "can_read"
            },
            "resource": {
                "id": 1, 
                "name": "DAG"
            }
        }
    ]
}

List Response Format

{
    "users": [...],
    "total_entries": 150,
    "count": 20
}

Types

from typing import Dict, List, Any, Optional
from flask import Blueprint, request, jsonify

# Request/Response types
UserDict = Dict[str, Any]
RoleDict = Dict[str, Any] 
PermissionDict = Dict[str, Any]
ResponseDict = Dict[str, Any]

Authentication

All API endpoints require authentication using one of the configured authentication backends:

  • Basic Auth: Authorization: Basic <base64-credentials>
  • Session Auth: Valid Flask session cookie
  • Kerberos Auth: Authorization: Negotiate <kerberos-token>

Base URL

The API endpoints are available at the base path: /auth/fab/v1/

Example full URLs:

  • GET /auth/fab/v1/users
  • POST /auth/fab/v1/users
  • GET /auth/fab/v1/roles
  • GET /auth/fab/v1/permissions

Install with Tessl CLI

npx tessl i tessl/pypi-apache-airflow-providers-fab

docs

api-endpoints.md

auth-backends.md

auth-manager.md

cli-commands.md

index.md

models.md

tile.json