feat: implement user authentication and permission-based data filtering
Add user login system with role-based access control: - Admin users see all data and can filter by any manager - Regular users only see records where ManagerName matches their username New components: - BIPUsersDAO: user authentication and management - SessionManager: singleton session state management - LoginDialog: modal login UI for app startup - init_users.sql: initial user setup script Permission enforcement: - Material validation tab: hide manager filter for non-admin, force filter by current user - Material type management: hide filter UI for non-admin, filter at database level - Window title and status bar display current user info Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
172
db/bip_users_dao.py
Normal file
172
db/bip_users_dao.py
Normal file
@@ -0,0 +1,172 @@
|
||||
"""
|
||||
BIPUsers DAO - Data access object for user authentication and management
|
||||
"""
|
||||
from typing import Optional, Dict, Any, List
|
||||
from db.connection import get_connection
|
||||
|
||||
|
||||
class BIPUsersDAO:
|
||||
"""Data access object for BIPUsers table"""
|
||||
|
||||
def authenticate(self, username: str, password: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Authenticate a user with username and password
|
||||
|
||||
Args:
|
||||
username: The username to authenticate
|
||||
password: The password to verify
|
||||
|
||||
Returns:
|
||||
Dict with user info if authentication successful, None otherwise
|
||||
Returns: {id, username, user_type}
|
||||
"""
|
||||
sql = """
|
||||
SELECT [ID], [UserName], [UserType]
|
||||
FROM [dbo].[BIPUsers]
|
||||
WHERE [UserName] = ? AND [Password] = ?
|
||||
"""
|
||||
with get_connection() as db:
|
||||
results = db.execute_query(sql, (username, password))
|
||||
if results:
|
||||
return {
|
||||
'id': results[0]['ID'],
|
||||
'username': results[0]['UserName'],
|
||||
'user_type': results[0]['UserType']
|
||||
}
|
||||
return None
|
||||
|
||||
def get_all_users(self) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all users from the database
|
||||
|
||||
Returns:
|
||||
List of user dictionaries: [{id, username, user_type, create_time}]
|
||||
"""
|
||||
sql = """
|
||||
SELECT [ID], [UserName], [UserType], [CreateTime]
|
||||
FROM [dbo].[BIPUsers]
|
||||
ORDER BY [UserName]
|
||||
"""
|
||||
with get_connection() as db:
|
||||
results = db.execute_query(sql)
|
||||
return [
|
||||
{
|
||||
'id': row['ID'],
|
||||
'username': row['UserName'],
|
||||
'user_type': row['UserType'],
|
||||
'create_time': row['CreateTime']
|
||||
}
|
||||
for row in results
|
||||
]
|
||||
|
||||
def create_user(self, username: str, password: str, user_type: str) -> bool:
|
||||
"""
|
||||
Create a new user
|
||||
|
||||
Args:
|
||||
username: The username (must be unique)
|
||||
password: The password (plain text for internal tool)
|
||||
user_type: User type ('Admin', 'User', or 'Guest')
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
sql = """
|
||||
INSERT INTO [dbo].[BIPUsers] ([UserName], [Password], [UserType])
|
||||
VALUES (?, ?, ?)
|
||||
"""
|
||||
try:
|
||||
with get_connection() as db:
|
||||
db.execute_update(sql, (username, password, user_type))
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error creating user: {e}")
|
||||
return False
|
||||
|
||||
def update_user_type(self, username: str, user_type: str) -> bool:
|
||||
"""
|
||||
Update a user's type
|
||||
|
||||
Args:
|
||||
username: The username to update
|
||||
user_type: New user type ('Admin', 'User', or 'Guest')
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
sql = """
|
||||
UPDATE [dbo].[BIPUsers]
|
||||
SET [UserType] = ?
|
||||
WHERE [UserName] = ?
|
||||
"""
|
||||
try:
|
||||
with get_connection() as db:
|
||||
db.execute_update(sql, (user_type, username))
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error updating user type: {e}")
|
||||
return False
|
||||
|
||||
def update_password(self, username: str, new_password: str) -> bool:
|
||||
"""
|
||||
Update a user's password
|
||||
|
||||
Args:
|
||||
username: The username to update
|
||||
new_password: The new password (plain text for internal tool)
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
sql = """
|
||||
UPDATE [dbo].[BIPUsers]
|
||||
SET [Password] = ?
|
||||
WHERE [UserName] = ?
|
||||
"""
|
||||
try:
|
||||
with get_connection() as db:
|
||||
db.execute_update(sql, (new_password, username))
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error updating password: {e}")
|
||||
return False
|
||||
|
||||
def delete_user(self, username: str) -> bool:
|
||||
"""
|
||||
Delete a user
|
||||
|
||||
Args:
|
||||
username: The username to delete
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
sql = """
|
||||
DELETE FROM [dbo].[BIPUsers]
|
||||
WHERE [UserName] = ?
|
||||
"""
|
||||
try:
|
||||
with get_connection() as db:
|
||||
db.execute_update(sql, (username,))
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error deleting user: {e}")
|
||||
return False
|
||||
|
||||
def user_exists(self, username: str) -> bool:
|
||||
"""
|
||||
Check if a username already exists
|
||||
|
||||
Args:
|
||||
username: The username to check
|
||||
|
||||
Returns:
|
||||
True if username exists, False otherwise
|
||||
"""
|
||||
sql = """
|
||||
SELECT COUNT(*) as count FROM [dbo].[BIPUsers]
|
||||
WHERE [UserName] = ?
|
||||
"""
|
||||
with get_connection() as db:
|
||||
results = db.execute_query(sql, (username,))
|
||||
return results[0]['count'] > 0 if results else False
|
||||
@@ -194,6 +194,20 @@ class MaterialsToBeDeletedDAO:
|
||||
results = db.execute_query(sql)
|
||||
return [r['ManagerName'] for r in results if r.get('ManagerName')]
|
||||
|
||||
def get_records_by_manager(self, manager_name: str) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get all material records for a specific manager.
|
||||
|
||||
This is an alias for get_materials_by_manager for consistency.
|
||||
|
||||
Args:
|
||||
manager_name: Manager name
|
||||
|
||||
Returns:
|
||||
List of material records for the specified manager
|
||||
"""
|
||||
return self.get_materials_by_manager(manager_name)
|
||||
|
||||
def get_record_by_material_code(self, material_code: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get a specific record by material code.
|
||||
|
||||
Reference in New Issue
Block a user