class User
A user management class that handles authentication, authorization, user profiles, preferences, file management, and logging for a Panel-based web application with Neo4j backend.
/tf/active/vicechatdev/userclass.py
23 - 396
complex
Purpose
The User class manages user authentication (both OAuth and local login), role-based access control (admin, manager, pathologist, read-only), user preferences, file operations, activity logging, and task completion tracking. It integrates with Neo4j graph database for user data storage and Panel framework for UI components. The class handles the complete user lifecycle from login to session management, including user statistics, preferences, file downloads, and audit logging.
Source Code
class User(param.Parameterized):
"""
User class to keep track of current user, roles, and information.
"""
login_state=param.String(default="undefined",doc="Login action to undertake")
def __init__(self):
super().__init__()
self.init_connections()
self.user = 'SYSTEM'
#self.name = self.user
self.number = ''
self.mail = ''
self.customer = ''
self.usergroup = ''
self.UID = ''
self.prefs={}
self.is_admin=False
self.is_manager=False
self.is_pathologist=False
self.is_ro=False
self.log = logging.getLogger(self.__class__.__name__)
self.log.setLevel(logging.DEBUG)
self.log.addHandler(self.file_handler)
self.log.addHandler(self.stream_handler)
self.modal_content=pn.Column(align = "center")
print("pn state user is", pn.state.user)
o365_user=pn.state.user
if o365_user is not None:
if '@' in o365_user:
g_text=f"match (u:User) WHERE u.Mail =~ '(?i){o365_user}' return u"
else:
g_text=f"match (u:User) WHERE u.Name =~ '(?i){o365_user}' return u"
selected_user=self.graph.run(g_text).evaluate()
if not selected_user:
self.login_state="local"
else:
self.fill_user(selected_user)
self.login_state="active"
else:
self.login_state="local"
self.user_stats=pn.Column(align = "center", sizing_mode = "stretch_width",max_width=1140, max_height=600,css_classes=['box1'])
self.user_prefs=pn.Column(align = "center", sizing_mode = "stretch_width",max_width=1140, max_height=600,css_classes=['box1'])
self.user_files=pn.Column(align = "center", sizing_mode = "stretch_width",max_width=1140, max_height=600,css_classes=['box1'])
self.user_logs=pn.Column(align = "center", sizing_mode = "stretch_width",max_width=1140, max_height=600,css_classes=['box1'])
return
def create_log(self):
"""Create log object"""
print("creating log")
logger = self.get_existing_logger(self.user)
if logger:
self.log = logger
else:
self.log = logging.getLogger(self.user)
self.log.setLevel(logging.DEBUG)
self.log.addHandler(self.file_handler)
def get_existing_logger(self, loggername):
"""
Checks if a logger already exists for the current user
Parameters
----------
loggername : str
The name of the logger
Returns
-------
Either the matched logger, or Bool False if no matching logger was found
"""
for name in logging.root.manager.loggerDict:
if name == loggername:
return logging.getLogger(loggername)
return False
def init_connections(self):
self.graph=Graph(config.DB_ADDR,auth=config.DB_AUTH,name=config.DB_NAME)
return
@property
def formatter(self):
"""
The logging formatter - defines the information passed to the log file and the console.
"""
return logging.Formatter('[%(asctime)s: %(name)s.%(funcName)s:%(levelname)s:]:\t%(message)s')
@property
def file_handler(self):
"""
The logging file handler. This defines the file we write to, the level of logging and the format of the output
"""
file_handler = logging.FileHandler(f"./logs/{self.user}.log")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(self.formatter)
return file_handler
@property
def stream_handler(self):
"""
The console log handler. This defines the level of logging, and the format of the output.
"""
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)
stream_handler.setFormatter(self.formatter)
return stream_handler
def fill_user(self,selected_user):
self.user=selected_user['Name']
self.UID=selected_user['UID']
self.prefs=json.loads(selected_user.get('Preferences',"{}"))
self.log = logging.getLogger(self.__class__.__name__)
self.log.setLevel(logging.DEBUG)
self.log.addHandler(self.file_handler)
self.log.addHandler(self.stream_handler)
try:
self.focal=str(selected_user['Focal']).split('/')
except:
pass
g_text="match (u:User {UID:'"+self.UID+"'})--(g:Users {N:'Management'}) return u"
outnode=self.graph.run(g_text).evaluate()
if outnode:
self.is_manager=True
self.log.info(f"User {self.user} has logged in from IP address {pn.state.headers.get('X-Real-Ip')}")
self.usergroups = self.graph.run("MATCH (:User {UID:'"+selected_user['UID']+"'})-->(u:Usergroup) return collect(u.Name)").evaluate()
print("usergroups : ",self.usergroups)
self.usergroup = self.usergroups[0]
if 'Admins' in self.usergroups:
print("setting as admin")
self.is_admin=True
self.customer = 'Internal'
elif 'Pathologists' in self.usergroups:
print("setting as pathologist")
self.is_pathologist=True
self.customer = 'Internal'
# if 'RO_users' in self.usergroups:
else:
print("setting as RO user")
self.is_ro=True
if self.customer!='Internal':
try:
self.customer = self.graph.run("match (c:Customer)<--(:User {N:'"+self.user+"'}) return c.N").evaluate()
except Exception as e:
self.log.warning(f"User {self.user} not linked to any customer")
self.customer = ''
try:
self.mail = selected_user.get('Mail','')
except Exception as e:
self.log.warning(f"User {self.user} has no mail address")
self.mail = ''
try:
self.number = selected_user.get('Number','')
except Exception as e:
self.log.warning(f"User {self.user} has no phone number")
self.number = ''
self.create_log()
return
def save_prefs(self):
self.graph.run("match (u:User {UID:'"+self.UID+"'}) set u.Preferences='"+json.dumps(self.prefs)+"'")
return
def login_check(self,event):
# if not self.username.value == 'Archivist':
# pn.state.notifications.error("Regular login only available to archive account")
# return
if '@' in self.username.value:
g_text=f"match (u:User) WHERE u.Mail =~ '(?i){self.username.value}' return u"
else:
g_text=f"match (u:User) WHERE u.Name =~ '(?i){self.username.value}' return u"
try:
selected_user=self.graph.run(g_text).evaluate()
print("selected user - ",selected_user)
except Exception as e:
self.log.warning(f"Invalid username or email entered: {self.username.value} from IP address {pn.state.headers.get('X-Real-Ip')}")
error = pn.pane.HTML("""
<div class="error">Username or Email address not found<div>
""")
self.modal_content.clear()
self.modal_content.append(error)
return_bt=pn.widgets.Button(name="Retry",button_type="danger")
def callback(event):
self.login_screen()
return_bt.on_click(callback)
self.modal_content.append(return_bt)
return
#print(hashlib.sha256(self.password.value.encode('utf-8')).hexdigest(),selected_user['password'])
if hashlib.sha256(self.password.value.encode('utf-8')).hexdigest()!=selected_user['password']:
self.log.warning(f"Failed login attempt for user {self.username.value} from IP address {pn.state.headers.get('X-Real-Ip')}")
error = pn.pane.HTML("""
<div class="error">You entered a wrong password<div>
""")
return_bt=pn.widgets.Button(name="Retry",button_type="danger")
def callback(event):
self.login_screen()
return_bt.on_click(callback)
self.modal_content.clear()
self.modal_content.append(error)
self.modal_content.append(return_bt)
else:
self.fill_user(selected_user)
success=pn.pane.HTML("""
<div class="error">You entered a wrong password<div>
""")
self.modal_content.clear()
self.modal_content.append(success)
self.login_state="active"
return
def login_screen(self):
spacer = pn.Spacer(height=10, margin=0)
html_pane = pn.pane.HTML("""
<h1>Welcome</h1>
""")
self.username = pn.widgets.TextInput(name='Please log in', placeholder='Username or mail', css_classes=['CPath-Input2'], max_width=400, align='center')
self.password = pn.widgets.PasswordInput(placeholder='Password', css_classes=['CPath-Input2','watch-for-enter'], max_width=400, align='center')
button = pn.widgets.Button(name='Log in', css_classes=['CPathBtn2'], max_width=400, align='center')
button.on_click(self.login_check)
items = pn.Column(html_pane, self.username, self.password, button)
main = pn.Column(items, align = "center", sizing_mode = "stretch_width",
max_width=1240)
self.modal_content.clear()
self.modal_content=pn.Column(spacer, main, sizing_mode= "stretch_both")
return
def build_user_stats(self):
self.user_stats.clear()
self.user_stats.append("Name : "+self.user)
self.user_stats.append("Email : "+self.mail)
self.user_stats.append("Number : "+self.number)
self.user_stats.append("Groups : "+",".join(self.usergroups))
btn = pn.widgets.Button(name="Clear cookies", width=80)
btn.on_click(self.clear_cookies)
self.user_stats.append(btn)
return
def clear_cookies(self, event=None):
cookie = pn.pane.HTML("""
<script>
document.cookie = "access_token=; expires=Thu, 01 Jan 1970 00:00:00 UTC;";
document.cookie = "id_token=; expires=Thu, 01 Jan 1970 00:00:00 UTC;";
document.cookie = "user=; expires=Thu, 01 Jan 1970 00:00:00 UTC;";
</script>
""")
self.user_stats.append(cookie)
return
def build_user_prefs(self):
self.user_prefs.clear()
for key, value in self.prefs.items():
self.user_prefs.append(str(key)+" : "+str(value))
return
def update_file_btn(self,event):
selected_files=self.files_tabulator.selected_dataframe['filename'].values.tolist()
zip_labels=[self.user+"/"+f for f in selected_files]
home_folder="/tf/stores/AllFileStore/users/"+self.UID+"/"
new_zip = BytesIO()
try:
with ZipFile(self.download_zip, 'r') as old_archive:
with ZipFile(new_zip, 'w') as new_archive:
files_present=[f.filename for f in old_archive.filelist]
for item in old_archive.filelist:
if item.filename in zip_labels:
new_archive.writestr(item, old_archive.read(item.filename))
for l in zip_labels:
if not (l in files_present):
with open(home_folder+l.split('/')[-1],'r+b') as file_in:
file1 = ZipInfo(l)
new_archive.writestr(file1, file_in.read())
except:
with ZipFile(new_zip, 'w') as new_archive:
for l in zip_labels:
with open(home_folder+l.split('/')[-1],'r+b') as file_in:
file1 = ZipInfo(l)
new_archive.writestr(file1, file_in.read())
self.download_zip=new_zip
self.download_zip.seek(0)
if selected_files!=[]:
self.download_btn.disabled=False
self.download_btn.filename=self.user+"_"+datetime.datetime.now().strftime("%Y_%m_%d_%I_%M_%S")+".zip"
self.download_btn.file=self.download_zip
return
def delete_user_files(self,event):
selected_files=self.files_tabulator.selected_dataframe['filename'].values.tolist()
home_folder="/tf/stores/AllFileStore/users/"+self.UID+"/"
for f in selected_files:
os.remove(home_folder+f)
my_files=glob.glob("/tf/stores/AllFileStore/users/"+self.UID+"/*")
short_user_files=sorted([f.split('/')[-1] for f in my_files])
self.files_df=pd.DataFrame(columns=['filename'],data=short_user_files)
self.files_tabulator=pn.widgets.Tabulator(self.files_df,show_index=False,selectable='checkbox', pagination='remote',max_height=500,page_size=10)
self.files_watcher=self.files_tabulator.param.watch(self.update_file_btn,['selection'],onlychanged=True)
self.user_files[1]=self.files_tabulator
return
def build_user_files(self):
my_files=glob.glob("/tf/stores/AllFileStore/users/"+self.UID+"/*")
short_user_files=sorted([f.split('/')[-1] for f in my_files])
self.user_files.clear()
#html_pane = pn.pane.HTML("""
#<h1 style="color: #aa64cf; margin: 20px 20px 20px 40px; text-align: center;"> My Files</h1>
#""")
#self.user_files.append(html_pane)
self.files_df=pd.DataFrame(columns=['filename'],data=short_user_files)
self.files_tabulator=pn.widgets.Tabulator(self.files_df,show_index=False,selectable='checkbox', pagination='remote',max_height=500,page_size=10)
self.files_watcher=self.files_tabulator.param.watch(self.update_file_btn,['selection'],onlychanged=True)
delete_btn=pn.widgets.Button(name="Delete selection",button_type='danger',width=100)
delete_btn.on_click(self.delete_user_files)
self.download_zip=BytesIO()
self.files_in_zip=[]
self.download_btn=pn.widgets.FileDownload(name="",filename="- Select your files first",file=self.download_zip, disabled=True)
self.user_files.append(pn.Row(self.download_btn))
self.user_files.append(self.files_tabulator)
return
def build_user_logs(self):
self.user_logs.clear()
#html_pane = pn.pane.HTML("""
#<h1 style="color: #aa64cf; margin: 20px 20px 20px 40px; text-align: center;"> My Logs</h1>
#""")
#self.user_logs.append(html_pane)
with open("./logs/"+self.user+".log") as f:
for line in f.readlines():
self.user_logs.insert(0,line)
return
def complete_task(self, object_UID, task, status="Completed", comment = "", delete = True, task_UID = None, **kwargs):
"""
Program function for creation of a stamp and deletion of the task object
Uses JSON format
"""
self.log.info(f"Updating task {task} for object with UID: '{object_UID}'")
self.log.info(f"The following kwargs were entered:\n{kwargs}")
task_dict = {"Name": task,
"Status":status,
"Stamp":f"Completed by {self.user} on {datetime.datetime.now().strftime('%Y-%m-%d at %H:%M')}",
"Comment":comment,
**kwargs}
self.log.debug(f"The task dict looks as follows:\n{task_dict}")
node = self.graph.run(f"MATCH (n) WHERE n.UID = '{object_UID}' return n").evaluate()
stamp = node.get('Stamp', '')
if stamp:
try: #Try for backwards compatibility > shouldn't exist tho
self.log.debug(f"Loading existing json to dict, json:\n{stamp}")
stamp_dict = json.loads(stamp)
except:
try:
stamp_dict = {'Old Stamp':stamp,
'Tasks':[]}
except Exception as e:
self.log.exception(e)
return
else:
stamp_dict = {"Tasks":[]}
stamp_dict['Tasks'].append(task_dict)
newstamp = json.dumps(stamp_dict)
self.graph.run(f"MATCH (n) WHERE n.UID = '{object_UID}' SET n.Stamp = '{newstamp}'")
self.log.info(f"Updated task {task} for object with UID: '{object_UID}'")
if delete:
if task_UID:
self.graph.run(f"MATCH (n)-->(t:Task {{UID:'{task_UID}'}}) WHERE n.UID = '{object_UID}' DETACH DELETE t")
else:
self.graph.run(f"MATCH (n)-->(t:Task {{T:'{task}'}}) WHERE n.UID = '{object_UID}' DETACH DELETE t")
Parameters
| Name | Type | Default | Kind |
|---|---|---|---|
bases |
param.Parameterized | - |
Parameter Details
__init__: No parameters required. Initializes a User instance with default SYSTEM user, establishes database connections, checks for OAuth user from pn.state.user, and sets up logging infrastructure. Automatically determines login state based on OAuth availability and user existence in database.
Return Value
Instantiation returns a User object with initialized state. Key method returns: get_existing_logger returns Logger object or False; fill_user returns None but populates user attributes; save_prefs returns None; login_check returns None but updates login_state; complete_task returns None but updates database with task completion stamps.
Class Interface
Methods
__init__(self) -> None
Purpose: Initializes User instance, establishes database connection, checks for OAuth user, sets up logging, and creates UI components
Returns: None
create_log(self) -> None
Purpose: Creates or retrieves existing logger for the current user with file and stream handlers
Returns: None
get_existing_logger(self, loggername: str) -> logging.Logger | bool
Purpose: Checks if a logger already exists for the given name
Parameters:
loggername: The name of the logger to search for
Returns: Logger object if found, False otherwise
init_connections(self) -> None
Purpose: Initializes Neo4j graph database connection using config settings
Returns: None
@property formatter(self) -> logging.Formatter
property
Purpose: Returns the logging formatter with timestamp, name, function, level, and message
Returns: Configured logging.Formatter instance
@property file_handler(self) -> logging.FileHandler
property
Purpose: Returns file handler for logging to ./logs/{username}.log
Returns: Configured logging.FileHandler instance
@property stream_handler(self) -> logging.StreamHandler
property
Purpose: Returns console stream handler for logging output
Returns: Configured logging.StreamHandler instance
fill_user(self, selected_user: dict) -> None
Purpose: Populates user attributes from Neo4j user node, determines roles, and sets up logging
Parameters:
selected_user: Dictionary containing user node properties from Neo4j (Name, UID, Preferences, Mail, Number, etc.)
Returns: None, but populates instance attributes: user, UID, prefs, focal, usergroups, usergroup, is_admin, is_manager, is_pathologist, is_ro, customer, mail, number
save_prefs(self) -> None
Purpose: Saves user preferences dictionary to Neo4j as JSON string
Returns: None
login_check(self, event) -> None
Purpose: Validates username/email and password against Neo4j database, updates login state and modal content
Parameters:
event: Panel button click event (not used directly)
Returns: None, but updates login_state to 'active' on success and populates modal_content with error or success messages
login_screen(self) -> None
Purpose: Builds and displays login UI with username, password inputs and login button in modal_content
Returns: None, but populates modal_content with login form
build_user_stats(self) -> None
Purpose: Builds user statistics panel showing name, email, number, groups, and cookie clear button
Returns: None, but populates user_stats Panel Column
clear_cookies(self, event=None) -> None
Purpose: Injects JavaScript to clear OAuth cookies (access_token, id_token, user)
Parameters:
event: Optional Panel button click event
Returns: None, but appends cookie-clearing HTML to user_stats
build_user_prefs(self) -> None
Purpose: Builds user preferences panel displaying all key-value pairs from prefs dictionary
Returns: None, but populates user_prefs Panel Column
update_file_btn(self, event) -> None
Purpose: Updates download button with selected files packaged in a ZIP archive
Parameters:
event: Panel Tabulator selection change event
Returns: None, but updates download_btn with new ZIP file and enables it
delete_user_files(self, event) -> None
Purpose: Deletes selected files from user's directory and refreshes file list
Parameters:
event: Panel button click event
Returns: None, but removes files from filesystem and updates files_tabulator
build_user_files(self) -> None
Purpose: Builds file management panel with tabulator showing user files, download and delete buttons
Returns: None, but populates user_files Panel Column with file management UI
build_user_logs(self) -> None
Purpose: Builds log viewer panel displaying user's log file in reverse chronological order
Returns: None, but populates user_logs Panel Column with log entries
complete_task(self, object_UID: str, task: str, status: str = 'Completed', comment: str = '', delete: bool = True, task_UID: str = None, **kwargs) -> None
Purpose: Marks a task as complete by adding a stamp to the object node and optionally deleting the task node
Parameters:
object_UID: UID of the Neo4j node to update with task completiontask: Name of the task being completedstatus: Status of task completion (default: 'Completed')comment: Optional comment about task completiondelete: Whether to delete the task node after completion (default: True)task_UID: Optional specific UID of task node to deletekwargs: Additional key-value pairs to include in task stamp
Returns: None, but updates object node's Stamp property with JSON task data and optionally deletes task node
Attributes
| Name | Type | Description | Scope |
|---|---|---|---|
login_state |
param.String | Current login state: 'undefined' (initial), 'local' (needs login), or 'active' (logged in) | class |
graph |
Graph | Neo4j graph database connection instance | instance |
user |
str | Username of current user (default: 'SYSTEM') | instance |
number |
str | Phone number of user | instance |
mail |
str | Email address of user | instance |
customer |
str | Customer organization user belongs to | instance |
usergroup |
str | Primary usergroup name | instance |
usergroups |
list | List of all usergroup names user belongs to | instance |
UID |
str | Unique identifier for user in database | instance |
prefs |
dict | User preferences dictionary loaded from JSON | instance |
is_admin |
bool | True if user is in Admins usergroup | instance |
is_manager |
bool | True if user is linked to Management usergroup | instance |
is_pathologist |
bool | True if user is in Pathologists usergroup | instance |
is_ro |
bool | True if user is read-only (not admin or pathologist) | instance |
log |
logging.Logger | Logger instance for user activity logging | instance |
modal_content |
pn.Column | Panel Column for displaying modal content (login screens, errors) | instance |
user_stats |
pn.Column | Panel Column displaying user statistics and information | instance |
user_prefs |
pn.Column | Panel Column displaying user preferences | instance |
user_files |
pn.Column | Panel Column for file management interface | instance |
user_logs |
pn.Column | Panel Column displaying user activity logs | instance |
username |
pn.widgets.TextInput | Login form username input widget | instance |
password |
pn.widgets.PasswordInput | Login form password input widget | instance |
files_tabulator |
pn.widgets.Tabulator | Tabulator widget displaying user files with selection | instance |
files_df |
pd.DataFrame | DataFrame containing user filenames | instance |
download_btn |
pn.widgets.FileDownload | File download button for selected files | instance |
download_zip |
BytesIO | In-memory ZIP file containing selected user files | instance |
focal |
list | List of focal points split from user's Focal property | instance |
Dependencies
loggingneo4j_driverparampanelhashlibglobpandasiodatetimeoszipfilejsonstringsecretsconfig
Required Imports
import logging
from neo4j_driver import *
import param
import panel as pn
import hashlib
import glob
import pandas as pd
from io import BytesIO
import datetime
import os
from zipfile import ZipFile, ZipInfo
import json
import config
Usage Example
# Initialize user (typically done once per session)
user = User()
# Check login state
if user.login_state == 'local':
# Show login screen
user.login_screen()
# Display modal_content in Panel app
# After login, user.login_state becomes 'active'
# Access user information
print(f"User: {user.user}")
print(f"Email: {user.mail}")
print(f"Is Admin: {user.is_admin}")
# Work with preferences
user.prefs['theme'] = 'dark'
user.save_prefs()
# Build UI components
user.build_user_stats()
user.build_user_files()
user.build_user_logs()
# Complete a task
user.complete_task(
object_UID='obj123',
task='Review',
status='Completed',
comment='All checks passed',
reviewer=user.user
)
# Access UI components
stats_panel = user.user_stats
files_panel = user.user_files
Best Practices
- Always check login_state before accessing user-specific features ('undefined', 'local', or 'active')
- Call init_connections() if database connection is lost to re-establish Graph connection
- Use save_prefs() after modifying user.prefs dictionary to persist changes to database
- Passwords are stored as SHA256 hashes; use hashlib.sha256(password.encode('utf-8')).hexdigest() for comparison
- User files are stored in /tf/stores/AllFileStore/users/{UID}/ directory structure
- Log files are created per user in ./logs/{username}.log
- The class uses Panel reactive components; ensure Panel server is running for UI features
- Role flags (is_admin, is_manager, is_pathologist, is_ro) are mutually exclusive based on usergroup membership
- Task completion uses JSON-formatted stamps stored in node.Stamp property
- File operations (download, delete) work with selected rows in files_tabulator widget
- Always handle exceptions when accessing user properties that may not exist (mail, number, customer)
- The modal_content attribute should be displayed in a Panel modal for login screens
- User preferences are stored as JSON strings in Neo4j and parsed to dict on load
Tags
Similar Components
AI-powered semantic similarity - components with related functionality:
-
class ControlledDocApp 56.9% similar
-
class UserCredential 56.0% similar
-
class CDocsApp 55.7% similar
-
class options 55.3% similar
-
class UserProfile_v1 53.6% similar