Apache Airflow Providers FAB

from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager

from airflow.providers.fab.auth_manager.models import User, Role, Permission, Action, Resource

from airflow.providers.fab.auth_manager.api.auth.backend import basic_auth, kerberos_auth, session

from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
from airflow.providers.fab.auth_manager.models import User, Role

# The FabAuthManager is typically configured through Airflow's configuration
# and used internally by Airflow's authentication system
auth_manager = FabAuthManager()

# Check if user is authorized for a specific action
if auth_manager.is_logged_in():
    user = auth_manager.get_user()
    print(f"Current user: {user.get_full_name()}")

class FabAuthManager(BaseAuthManager):
    def get_user(self) -> User: ...
    def is_logged_in(self) -> bool: ...
    def is_authorized_dag(self, *, method: ResourceMethod, access_entity: DagAccessEntity | None = None, details: DagDetails | None = None, user: BaseUser | None = None) -> bool: ...
    def is_authorized_view(self, *, access_view: AccessView, user: BaseUser | None = None) -> bool: ...
    def get_permitted_dag_ids(self, *, methods: Container[ResourceMethod] | None = None, user: BaseUser | None = None, session: Session = NEW_SESSION) -> set[str]: ...

class User(Model, BaseUser):
    id: int
    username: str
    email: str
    first_name: str
    last_name: str
    active: bool
    roles: list[Role]
    
    def get_full_name(self) -> str: ...
    def get_id(self) -> int: ...

class Role(Model):
    id: int
    name: str
    permissions: list[Permission]
    
class Permission(Model):
    id: int
    action: Action
    resource: Resource

def users_create(username: str, email: str, firstname: str, lastname: str, role: str, password: str | None = None, use_random_password: bool = False) -> None: ...
def roles_create(roles: list[str]) -> None: ...
def sync_perm(include_dags: bool = False) -> None: ...

# Basic Auth Backend
def auth_current_user() -> User | None: ...
def requires_authentication(function: T) -> T: ...

# Kerberos Auth Backend  
class KerberosService: ...
def find_user(username: str | None = None, email: str | None = None) -> User | None: ...

# Session Auth Backend
def requires_authentication(function: T) -> T: ...

# User endpoint operations
def get_users() -> dict: ...
def get_user(user_id: int) -> dict: ...
def patch_user(user_id: int, data: dict) -> dict: ...

# Role and permission endpoint operations  
def get_roles() -> dict: ...
def get_role(role_id: int) -> dict: ...
def get_permissions() -> dict: ...

from typing import TypeVar, Callable, Sequence, Container
from airflow.auth.managers.models.base_user import BaseUser
from airflow.auth.managers.base_auth_manager import ResourceMethod
from airflow.auth.managers.models.resource_details import (
    DagDetails,
    DagAccessEntity,
    AccessView,
    ConnectionDetails,
    VariableDetails,
    PoolDetails,
    ConfigurationDetails
)
from sqlalchemy.orm import Session
from airflow.utils.session import NEW_SESSION

T = TypeVar("T", bound=Callable)