Auth Manager

class FabAuthManager(BaseAuthManager):
    @staticmethod
    def get_cli_commands() -> list[CLICommand]:
        """Returns CLI commands to be included in Airflow CLI."""
        
    def get_api_endpoints(self) -> Blueprint | None:
        """Returns Flask Blueprint for API endpoints."""
        
    def get_user_display_name(self) -> str:
        """Returns the user's display name associated with the session."""
        
    def get_user(self) -> User:
        """Returns the user associated with the current session."""
        
    def init(self) -> None:
        """Initializes the auth manager."""
        
    def is_logged_in(self) -> bool:
        """Checks if the current user is authenticated."""

def is_authorized_configuration(
    self, 
    *, 
    method: ResourceMethod,
    details: ConfigurationDetails | None = None, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access configuration."""

def is_authorized_connection(
    self, 
    *, 
    method: ResourceMethod,
    details: ConnectionDetails | None = None, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access connection."""

def is_authorized_dag(
    self, 
    *, 
    method: ResourceMethod,
    access_entity: DagAccessEntity | None = None,
    details: DagDetails | None = None, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access DAG."""

def is_authorized_asset(
    self, 
    *, 
    method: ResourceMethod,
    details: AssetDetails | None = None, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access asset."""

def is_authorized_dataset(
    self, 
    *, 
    method: ResourceMethod,
    details: AssetDetails | None = None, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access dataset.
    
    .. deprecated:: Airflow 3.0.0
        Use `is_authorized_asset` instead. This method will be removed in a future version.
    """

def is_authorized_pool(
    self, 
    *, 
    method: ResourceMethod,
    details: PoolDetails | None = None, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access pool."""

def is_authorized_variable(
    self, 
    *, 
    method: ResourceMethod,
    details: VariableDetails | None = None, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access variable."""

def is_authorized_view(
    self, 
    *, 
    access_view: AccessView, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access view."""

def is_authorized_custom_view(
    self, 
    *, 
    method: ResourceMethod | str, 
    resource_name: str, 
    user: BaseUser | None = None
) -> bool:
    """Checks if user is authorized to access custom view."""

def get_permitted_dag_ids(
    self, 
    *, 
    methods: Container[ResourceMethod] | None = None,
    user: BaseUser | None = None,
    session: Session = NEW_SESSION
) -> set[str]:
    """Returns set of DAG IDs that the user has permission to access."""

@cached_property
def security_manager(self) -> FabAirflowSecurityManagerOverride:
    """Returns the security manager instance."""

def get_url_login(self, **kwargs) -> str:
    """Returns the login URL with optional parameters."""

def get_url_logout(self) -> str:
    """Returns the logout URL."""
    
def register_views(self) -> None:
    """Registers views with the application."""

def get_url_user_profile(self) -> str | None:
    """Returns the user profile URL."""

from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
from airflow.auth.managers.models.resource_details import DagDetails
from airflow.auth.managers.base_auth_manager import ResourceMethod

auth_manager = FabAuthManager()

# Check if user is logged in
if auth_manager.is_logged_in():
    user = auth_manager.get_user()
    print(f"User: {user.get_full_name()}")
    
    # Check DAG access
    dag_details = DagDetails(id="my_dag")
    if auth_manager.is_authorized_dag(method=ResourceMethod.GET, details=dag_details):
        print("User has access to my_dag")

# Get DAGs user has access to
permitted_dags = auth_manager.get_permitted_dag_ids(methods=[ResourceMethod.GET, ResourceMethod.PUT])
print(f"User can access: {permitted_dags}")

# Generate authentication URLs
login_url = auth_manager.get_url_login(next="/admin/")
logout_url = auth_manager.get_url_logout()
profile_url = auth_manager.get_url_user_profile()

print(f"Login: {login_url}")
print(f"Logout: {logout_url}")
print(f"Profile: {profile_url}")

from typing import TYPE_CHECKING, Container
from functools import cached_property
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, AccessView, ConnectionDetails, VariableDetails, 
    PoolDetails, ConfigurationDetails, DagAccessEntity
)
from airflow.cli.cli_config import CLICommand
from flask import Blueprint
from sqlalchemy.orm import Session
from airflow.utils.session import NEW_SESSION

if TYPE_CHECKING:
    from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
    from airflow.providers.common.compat.assets import AssetDetails