Models

class User(Model, BaseUser):
    """Represents an Airflow user which has roles assigned to it."""
    
    # Database fields
    id: int  # Primary key
    first_name: str  # User's first name (max 256 chars)
    last_name: str   # User's last name (max 256 chars)
    username: str    # Unique username (max 512 chars)
    password: str    # Hashed password (max 256 chars)
    active: bool     # Whether user account is active
    email: str       # Unique email address (max 512 chars)
    last_login: datetime.datetime | None     # Last login timestamp
    login_count: int | None                  # Number of successful logins
    fail_login_count: int | None             # Number of failed login attempts
    roles: list[Role]                        # Associated roles
    created_on: datetime.datetime | None     # Account creation timestamp
    changed_on: datetime.datetime | None     # Last modification timestamp
    created_by_fk: int | None                # ID of user who created this account
    changed_by_fk: int | None                # ID of user who last modified this account
    
    # Relationships
    created_by: User | None     # User who created this account
    changed_by: User | None     # User who last modified this account
    
    # Properties
    @property
    def is_authenticated(self) -> bool:
        """Always returns True for authenticated users."""
        
    @property
    def is_active(self) -> bool:
        """Returns the active status of the user."""
        
    @property
    def is_anonymous(self) -> bool:
        """Always returns False for registered users."""
        
    @property
    def perms(self) -> set[tuple[str, str]]:
        """Returns set of permission tuples (action, resource) for the user."""
    
    # Methods
    def get_id(self) -> int:
        """Returns the user's ID."""
        
    def get_name(self) -> str:
        """Returns username, email, or user_id as identifier."""
        
    def get_full_name(self) -> str:
        """Returns formatted full name."""
        
    @classmethod
    def get_user_id(cls) -> int | None:
        """Returns current user ID from Flask global context."""

class Role(Model):
    """Represents a user role to which permissions can be assigned."""
    
    id: int           # Primary key
    name: str         # Unique role name (max 64 chars)
    permissions: list[Permission]  # Associated permissions

class Permission(Model):
    """Permission pair comprised of an Action + Resource combo."""
    
    id: int                    # Primary key
    action_id: int             # Foreign key to Action
    action: Action             # Action relationship
    resource_id: int           # Foreign key to Resource
    resource: Resource         # Resource relationship

class Action(Model):
    """Represents permission actions such as `can_read`."""
    
    id: int      # Primary key
    name: str    # Unique action name (max 100 chars)

class Resource(Model):
    """Represents permission object such as `User` or `Dag`."""
    
    id: int      # Primary key
    name: str    # Unique resource name (max 250 chars)
    
    def __eq__(self, other) -> bool:
        """Equality comparison based on name."""
        
    def __neq__(self, other) -> bool:
        """Inequality comparison based on name."""

class RegisterUser(Model):
    """Represents a user registration."""
    
    id: int                              # Primary key
    first_name: str                      # First name (max 256 chars)
    last_name: str                       # Last name (max 256 chars)
    username: str                        # Unique username (max 512 chars)
    password: str                        # Hashed password (max 256 chars) 
    email: str                           # Email address (max 512 chars)
    registration_date: datetime.datetime | None  # Registration timestamp
    registration_hash: str               # Registration verification hash (max 256 chars)

from airflow.providers.fab.auth_manager.models import User, Role
from sqlalchemy.orm import Session

# Query users
with Session() as session:
    # Get all active users
    active_users = session.query(User).filter(User.active == True).all()
    
    # Get user by username
    user = session.query(User).filter(User.username == "admin").first()
    if user:
        print(f"User: {user.get_full_name()}")
        print(f"Email: {user.email}")
        print(f"Active: {user.is_active}")
        print(f"Roles: {[role.name for role in user.roles]}")

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

# Get user permissions
user = session.query(User).filter(User.username == "analyst").first()
if user:
    # Get all permissions for user
    user_perms = user.perms
    print(f"User permissions: {user_perms}")
    
    # Check specific permission
    has_dag_read = ("can_read", "DAG") in user_perms
    print(f"Can read DAGs: {has_dag_read}")

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

# Query roles and their permissions
admin_role = session.query(Role).filter(Role.name == "Admin").first()
if admin_role:
    print(f"Role: {admin_role.name}")
    for perm in admin_role.permissions:
        print(f"  {perm.action.name} on {perm.resource.name}")

# Check user authentication status
if user.is_authenticated and user.is_active and not user.is_anonymous:
    print(f"User {user.get_name()} is properly authenticated")
    print(f"Login count: {user.login_count}")
    print(f"Last login: {user.last_login}")

import datetime
from typing import TYPE_CHECKING
from flask_appbuilder.models.sqla import Model
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from airflow.auth.managers.models.base_user import BaseUser

if TYPE_CHECKING:
    from sqlalchemy import Identity