tessl/pypi-apache-airflow-fab-security
Flask-AppBuilder (FAB) security integration component within Apache Airflow core, providing authentication, authorization, and security management features
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
user-management.mddocs/
User Management
Comprehensive user lifecycle management including creation, updates, deletion, and queries. Handles user authentication statistics, profile management, and user registration workflows.
Capabilities
User Creation and Registration
Create new users with role assignments and handle user registration requests.
def add_user(
self,
username: str,
first_name: str,
last_name: str,
email: str,
role: Role | list[Role],
password: str = "",
hashed_password: str = ""
) -> User | None:
"""
Create a new user account.
Parameters:
- username: Unique username for the user
- first_name: User's first name
- last_name: User's last name
- email: User's email address (must be unique)
- role: Role object or list of roles to assign to user
- password: Plain text password (will be hashed if provided)
- hashed_password: Pre-hashed password (use instead of password)
Returns:
User object if created successfully, None if creation failed
"""
def add_register_user(
self,
username: str,
first_name: str,
last_name: str,
email: str,
password: str = "",
hashed_password: str = ""
) -> RegisterUser | None:
"""
Add a user registration request.
Parameters:
- username: Requested username
- first_name: User's first name
- last_name: User's last name
- email: User's email address
- password: Plain text password (will be hashed)
- hashed_password: Pre-hashed password (use instead of password)
Returns:
RegisterUser object if registration added successfully, None otherwise
"""User Queries and Retrieval
Find and retrieve users by various criteria including ID, username, and email.
def find_user(self, username: str = None, email: str = None) -> User | None:
"""
Find user by username or email address.
Parameters:
- username: Username to search for (case-insensitive if configured)
- email: Email address to search for
Returns:
User object if found, None otherwise
"""
def get_user_by_id(self, pk: int) -> User | None:
"""
Get user by primary key ID.
Parameters:
- pk: User's primary key ID
Returns:
User object if found, None otherwise
"""
def get_all_users(self) -> list[User]:
"""
Get all users in the system.
Returns:
List of all User objects
"""
def count_users(self) -> int:
"""
Count total number of users in the system.
Returns:
Total user count as integer
"""User Updates and Maintenance
Update existing user information and manage user accounts.
def update_user(self, user: User) -> bool:
"""
Update user information in the database.
Parameters:
- user: User object with updated information
Returns:
True if update successful, False otherwise
"""Registration Management
Manage user registration requests and approvals.
def find_register_user(self, registration_hash: str) -> RegisterUser | None:
"""
Find user registration by hash.
Parameters:
- registration_hash: Registration hash string
Returns:
RegisterUser object if found, None otherwise
"""
def del_register_user(self, register_user: RegisterUser) -> bool:
"""
Delete user registration request.
Parameters:
- register_user: RegisterUser object to delete
Returns:
True if deletion successful, False otherwise
"""User Data Models
Access to user data model interfaces for advanced operations.
@property
def get_user_datamodel(self):
"""Get the User data model interface."""
@property
def get_register_user_datamodel(self):
"""Get the RegisterUser data model interface."""Usage Examples
Creating Users
from airflow.www.fab_security.sqla.manager import SecurityManager
# Create a new user with a role
user_role = security_manager.find_role("User")
new_user = security_manager.add_user(
username="jane_doe",
first_name="Jane",
last_name="Doe",
email="jane@example.com",
role=user_role,
password="secure_password123"
)
if new_user:
print(f"User created: {new_user.get_full_name()}")Finding Users
# Find user by username
user = security_manager.find_user(username="jane_doe")
if user:
print(f"Found user: {user.email}")
# Find user by email
user = security_manager.find_user(email="jane@example.com")
# Get user by ID
user = security_manager.get_user_by_id(1)User Registration Workflow
# Add registration request
registration = security_manager.add_register_user(
username="new_user",
first_name="New",
last_name="User",
email="new@example.com",
password="temp_password"
)
if registration:
print(f"Registration hash: {registration.registration_hash}")
# Later, find and process registration
found_reg = security_manager.find_register_user(registration.registration_hash)
if found_reg:
# Create actual user account
user = security_manager.add_user(
username=found_reg.username,
first_name=found_reg.first_name,
last_name=found_reg.last_name,
email=found_reg.email,
role=security_manager.find_role("User"),
hashed_password=found_reg.password # Already hashed
)
# Clean up registration
security_manager.del_register_user(found_reg)Updating User Information
# Find and update user
user = security_manager.find_user(username="jane_doe")
if user:
user.first_name = "Janet"
user.email = "janet@example.com"
success = security_manager.update_user(user)
if success:
print("User updated successfully")User Statistics
# Get total user count
total_users = security_manager.count_users()
print(f"Total users: {total_users}")
# Get all users for administrative purposes
all_users = security_manager.get_all_users()
for user in all_users:
print(f"User: {user.username} ({user.email})")Error Handling
User management methods handle errors gracefully:
add_user()returnsNoneon failure and logs the errorfind_user()returnsNoneif user not foundupdate_user()returnsFalseon failure- Database constraints (unique username/email) are enforced
- Invalid role assignments are handled with appropriate logging
Install with Tessl CLI
npx tessl i tessl/pypi-apache-airflow-fab-security