0
# User Management
1

2
Comprehensive user lifecycle management including creation, updates, deletion, and queries. Handles user authentication statistics, profile management, and user registration workflows.
3

4
## Capabilities
5

6
### User Creation and Registration
7

8
Create new users with role assignments and handle user registration requests.
9

10
```python { .api }
11
def add_user(
12
    self,
13
    username: str,
14
    first_name: str,
15
    last_name: str,
16
    email: str,
17
    role: Role | list[Role],
18
    password: str = "",
19
    hashed_password: str = ""
20
) -> User | None:
21
    """
22
    Create a new user account.
23
    
24
    Parameters:
25
    - username: Unique username for the user
26
    - first_name: User's first name
27
    - last_name: User's last name
28
    - email: User's email address (must be unique)
29
    - role: Role object or list of roles to assign to user
30
    - password: Plain text password (will be hashed if provided)
31
    - hashed_password: Pre-hashed password (use instead of password)
32
    
33
    Returns:
34
    User object if created successfully, None if creation failed
35
    """
36

37
def add_register_user(
38
    self,
39
    username: str,
40
    first_name: str,
41
    last_name: str,
42
    email: str,
43
    password: str = "",
44
    hashed_password: str = ""
45
) -> RegisterUser | None:
46
    """
47
    Add a user registration request.
48
    
49
    Parameters:
50
    - username: Requested username
51
    - first_name: User's first name
52
    - last_name: User's last name
53
    - email: User's email address
54
    - password: Plain text password (will be hashed)
55
    - hashed_password: Pre-hashed password (use instead of password)
56
    
57
    Returns:
58
    RegisterUser object if registration added successfully, None otherwise
59
    """
60
```
61

62
### User Queries and Retrieval
63

64
Find and retrieve users by various criteria including ID, username, and email.
65

66
```python { .api }
67
def find_user(self, username: str = None, email: str = None) -> User | None:
68
    """
69
    Find user by username or email address.
70
    
71
    Parameters:
72
    - username: Username to search for (case-insensitive if configured)
73
    - email: Email address to search for
74
    
75
    Returns:
76
    User object if found, None otherwise
77
    """
78

79
def get_user_by_id(self, pk: int) -> User | None:
80
    """
81
    Get user by primary key ID.
82
    
83
    Parameters:
84
    - pk: User's primary key ID
85
    
86
    Returns:
87
    User object if found, None otherwise
88
    """
89

90
def get_all_users(self) -> list[User]:
91
    """
92
    Get all users in the system.
93
    
94
    Returns:
95
    List of all User objects
96
    """
97

98
def count_users(self) -> int:
99
    """
100
    Count total number of users in the system.
101
    
102
    Returns:
103
    Total user count as integer
104
    """
105
```
106

107
### User Updates and Maintenance
108

109
Update existing user information and manage user accounts.
110

111
```python { .api }
112
def update_user(self, user: User) -> bool:
113
    """
114
    Update user information in the database.
115
    
116
    Parameters:
117
    - user: User object with updated information
118
    
119
    Returns:
120
    True if update successful, False otherwise
121
    """
122
```
123

124
### Registration Management
125

126
Manage user registration requests and approvals.
127

128
```python { .api }
129
def find_register_user(self, registration_hash: str) -> RegisterUser | None:
130
    """
131
    Find user registration by hash.
132
    
133
    Parameters:
134
    - registration_hash: Registration hash string
135
    
136
    Returns:
137
    RegisterUser object if found, None otherwise
138
    """
139

140
def del_register_user(self, register_user: RegisterUser) -> bool:
141
    """
142
    Delete user registration request.
143
    
144
    Parameters:
145
    - register_user: RegisterUser object to delete
146
    
147
    Returns:
148
    True if deletion successful, False otherwise
149
    """
150
```
151

152
### User Data Models
153

154
Access to user data model interfaces for advanced operations.
155

156
```python { .api }
157
@property
158
def get_user_datamodel(self):
159
    """Get the User data model interface."""
160

161
@property
162
def get_register_user_datamodel(self):
163
    """Get the RegisterUser data model interface."""
164
```
165

166
## Usage Examples
167

168
### Creating Users
169

170
```python
171
from airflow.www.fab_security.sqla.manager import SecurityManager
172

173
# Create a new user with a role
174
user_role = security_manager.find_role("User")
175
new_user = security_manager.add_user(
176
    username="jane_doe",
177
    first_name="Jane",
178
    last_name="Doe",
179
    email="jane@example.com",
180
    role=user_role,
181
    password="secure_password123"
182
)
183

184
if new_user:
185
    print(f"User created: {new_user.get_full_name()}")
186
```
187

188
### Finding Users
189

190
```python
191
# Find user by username
192
user = security_manager.find_user(username="jane_doe")
193
if user:
194
    print(f"Found user: {user.email}")
195

196
# Find user by email
197
user = security_manager.find_user(email="jane@example.com")
198

199
# Get user by ID
200
user = security_manager.get_user_by_id(1)
201
```
202

203
### User Registration Workflow
204

205
```python
206
# Add registration request
207
registration = security_manager.add_register_user(
208
    username="new_user",
209
    first_name="New",
210
    last_name="User",
211
    email="new@example.com",
212
    password="temp_password"
213
)
214

215
if registration:
216
    print(f"Registration hash: {registration.registration_hash}")
217
    
218
    # Later, find and process registration
219
    found_reg = security_manager.find_register_user(registration.registration_hash)
220
    if found_reg:
221
        # Create actual user account
222
        user = security_manager.add_user(
223
            username=found_reg.username,
224
            first_name=found_reg.first_name,
225
            last_name=found_reg.last_name,
226
            email=found_reg.email,
227
            role=security_manager.find_role("User"),
228
            hashed_password=found_reg.password  # Already hashed
229
        )
230
        
231
        # Clean up registration
232
        security_manager.del_register_user(found_reg)
233
```
234

235
### Updating User Information
236

237
```python
238
# Find and update user
239
user = security_manager.find_user(username="jane_doe")
240
if user:
241
    user.first_name = "Janet"
242
    user.email = "janet@example.com"
243
    
244
    success = security_manager.update_user(user)
245
    if success:
246
        print("User updated successfully")
247
```
248

249
### User Statistics
250

251
```python
252
# Get total user count
253
total_users = security_manager.count_users()
254
print(f"Total users: {total_users}")
255

256
# Get all users for administrative purposes
257
all_users = security_manager.get_all_users()
258
for user in all_users:
259
    print(f"User: {user.username} ({user.email})")
260
```
261

262
## Error Handling
263

264
User management methods handle errors gracefully:
265
- `add_user()` returns `None` on failure and logs the error
266
- `find_user()` returns `None` if user not found
267
- `update_user()` returns `False` on failure
268
- Database constraints (unique username/email) are enforced
269
- Invalid role assignments are handled with appropriate logging