User Management

import chainlit as cl

class User:
    """
    User representation for authentication and session management.
    
    Args:
        identifier: str - Unique user identifier (email, username, etc.)
        display_name: Optional[str] - Human-readable display name
        metadata: Dict - Additional user data and custom properties
        
    Returns:
        User instance for authentication and session tracking
    """
    def __init__(
        self,
        identifier: str,
        display_name: Optional[str] = None,
        metadata: Dict = {}
    ): ...

class PersistedUser:
    """
    Extended User class with database persistence fields.
    
    Args:
        identifier: str - Unique user identifier
        display_name: Optional[str] - Human-readable display name  
        metadata: Dict - Additional user data
        id: str - Database record ID
        createdAt: str - Creation timestamp
        
    Returns:
        PersistedUser instance with database tracking
    """
    def __init__(
        self,
        identifier: str,
        display_name: Optional[str] = None,
        metadata: Dict = {},
        id: str = "",
        createdAt: str = ""
    ): ...

import chainlit as cl

# Create a basic user
user = cl.User(
    identifier="alice@example.com",
    display_name="Alice Smith",
    metadata={
        "role": "admin",
        "preferences": {"theme": "dark"},
        "last_login": "2024-01-15"
    }
)

# Access user properties
print(f"User: {user.display_name} ({user.identifier})")
print(f"Role: {user.metadata.get('role')}")

# Persisted user with database fields
persisted_user = cl.PersistedUser(
    identifier="bob@example.com",
    display_name="Bob Johnson",
    metadata={"department": "engineering"},
    id="user_12345",
    createdAt="2024-01-01T00:00:00Z"
)

class UserSession:
    """
    Session storage for user-specific data between interactions.
    Accessible via the global `user_session` object.
    
    Built-in fields:
        id: str - Session identifier
        env: Dict - Environment variables
        chat_settings: Dict - Current chat settings
        user: Optional[User] - Authenticated user object
        chat_profile: Optional[str] - Active chat profile
        client_type: str - Client type (web, mobile, etc.)
    """
    
    def get(self, key: str, default: Any = None) -> Any:
        """
        Retrieve a value from the session.
        
        Args:
            key: str - Session key to retrieve
            default: Any - Default value if key not found
            
        Returns:
            Stored value or default
        """
        
    def set(self, key: str, value: Any) -> None:
        """
        Store a value in the session.
        
        Args:
            key: str - Session key to store
            value: Any - Value to store
        """
        
    def create_accessor(
        self, 
        key: str, 
        default: Any = None, 
        apply_fn: Optional[Callable] = None
    ) -> Callable:
        """
        Create a typed accessor for a session key.
        
        Args:
            key: str - Session key
            default: Any - Default value for the key
            apply_fn: Optional[Callable] - Transform function for the value
            
        Returns:
            Callable accessor function
        """

# Global session object available in all callback contexts
user_session: UserSession

import chainlit as cl

@cl.on_chat_start
async def start():
    """Initialize user session data"""
    # Store user preferences
    cl.user_session.set("user_preferences", {
        "language": "en",
        "notifications": True,
        "theme": "light"
    })
    
    # Store conversation context
    cl.user_session.set("conversation_history", [])
    cl.user_session.set("current_topic", None)
    
    # Access built-in session fields
    user = cl.user_session.get("user")
    if user:
        await cl.Message(f"Welcome back, {user.display_name}!").send()
    else:
        await cl.Message("Welcome! Please sign in to continue.").send()

@cl.on_message
async def handle_message(message: cl.Message):
    # Retrieve session data
    preferences = cl.user_session.get("user_preferences", {})
    history = cl.user_session.get("conversation_history", [])
    
    # Update conversation history
    history.append({
        "timestamp": "2024-01-15T10:30:00Z",
        "message": message.content,
        "type": "user"
    })
    cl.user_session.set("conversation_history", history)
    
    # Use preferences in response
    language = preferences.get("language", "en")
    response = f"[{language}] Processing your message..."
    await cl.Message(response).send()
    
    # Track conversation state
    cl.user_session.set("last_interaction", "2024-01-15T10:30:00Z")

# Create typed accessors for common session data
@cl.on_chat_start  
async def setup_accessors():
    # Create accessor with default value
    get_user_score = cl.user_session.create_accessor("user_score", default=0)
    
    # Create accessor with transform function
    get_user_level = cl.user_session.create_accessor(
        "user_level", 
        default="beginner",
        apply_fn=str.lower
    )
    
    # Use accessors
    score = get_user_score()  # Returns 0 if not set
    level = get_user_level()  # Returns "beginner" and lowercases value
    
    await cl.Message(f"Your level: {level}, Score: {score}").send()

@cl.on_settings_update
async def handle_settings_update(settings: Dict[str, Any]):
    """Handle chat settings changes"""
    # Store updated settings in session
    cl.user_session.set("chat_settings", settings)
    
    # Access via built-in field
    current_settings = cl.user_session.chat_settings
    
    await cl.Message("Settings updated successfully!").send()

# Access session data across different callbacks
@cl.on_chat_resume
async def resume_chat(thread_dict: Dict):
    """Restore session state when resuming a chat"""
    # Restore conversation context
    cl.user_session.set("thread_id", thread_dict.get("id"))
    
    # Load user-specific data
    user = cl.user_session.get("user")
    if user:
        # Restore user preferences from metadata
        preferences = user.metadata.get("preferences", {})
        cl.user_session.set("user_preferences", preferences)
        
        await cl.Message(f"Resuming conversation, {user.display_name}").send()

@cl.on_chat_end
async def cleanup_session():
    """Clean up session data when chat ends"""
    # Save important session data before cleanup
    final_score = cl.user_session.get("user_score", 0)
    conversation_summary = cl.user_session.get("conversation_summary", "")
    
    # Store in user metadata for next session
    user = cl.user_session.get("user")
    if user:
        user.metadata.update({
            "last_score": final_score,
            "last_summary": conversation_summary
        })

@dataclass
class ChatProfile:
    """
    Chat profile configuration for different conversation modes.
    
    Fields:
        name: str - Profile identifier  
        markdown_description: str - Profile description in markdown
        icon: Optional[str] - Icon name for the profile
        default: bool - Whether this is the default profile
        starters: Optional[List[Starter]] - Profile-specific conversation starters
    """
    name: str
    markdown_description: str
    icon: Optional[str] = None
    default: bool = False
    starters: Optional[List[Starter]] = None

@dataclass
class Starter:
    """
    Conversation starter configuration for chat profiles.
    
    Fields:
        label: str - Display text for the starter
        message: str - Message content when starter is selected
        command: Optional[str] - Optional command identifier
        icon: Optional[str] - Icon name for the starter
    """
    label: str
    message: str
    command: Optional[str] = None
    icon: Optional[str] = None

import chainlit as cl

@cl.set_chat_profiles
async def chat_profiles(user: Optional[cl.User]) -> List[cl.ChatProfile]:
    """Define available chat profiles for users"""
    profiles = [
        cl.ChatProfile(
            name="assistant",
            markdown_description="General AI assistant for questions and tasks",
            icon="message-circle",
            default=True,
            starters=[
                cl.Starter(
                    label="Ask a question",
                    message="What would you like to know?",
                    icon="help-circle"
                ),
                cl.Starter(
                    label="Get help with coding",  
                    message="I need help with coding",
                    icon="code"
                )
            ]
        ),
        cl.ChatProfile(
            name="analyst", 
            markdown_description="Data analysis and visualization expert",
            icon="bar-chart-3",
            starters=[
                cl.Starter(
                    label="Analyze data",
                    message="Help me analyze my dataset",
                    icon="trending-up"
                ),
                cl.Starter(
                    label="Create visualization",
                    message="Create a chart from my data", 
                    icon="pie-chart"
                )
            ]
        )
    ]
    
    # Filter profiles based on user permissions
    if user and user.metadata.get("role") == "admin":
        profiles.append(
            cl.ChatProfile(
                name="admin",
                markdown_description="Administrative functions and system management",
                icon="settings",
                starters=[
                    cl.Starter(
                        label="System status",
                        message="Show system status",
                        icon="activity"
                    )
                ]
            )
        )
    
    return profiles

@cl.on_chat_start
async def handle_profile_start():
    """Handle chat initialization based on selected profile"""
    # Get current chat profile from session
    profile = cl.user_session.get("chat_profile")
    
    if profile == "assistant":
        await cl.Message("Hi! I'm your AI assistant. How can I help you today?").send()
    elif profile == "analyst":
        await cl.Message("Hello! I'm here to help with data analysis. Upload your data or ask me about analytics.").send()
    elif profile == "admin":
        await cl.Message("Admin mode activated. What would you like to manage?").send()
    else:
        await cl.Message("Welcome! Please select a chat profile to get started.").send()

@cl.on_message
async def handle_profile_message(message: cl.Message):
    """Handle messages based on active chat profile"""
    profile = cl.user_session.get("chat_profile", "assistant")
    
    if profile == "analyst":
        # Handle data analysis requests
        if "upload" in message.content.lower() or "data" in message.content.lower():
            files = await cl.AskFileMessage(
                content="Please upload your data file:",
                accept=["text/csv", "application/json", "text/plain"]
            ).send()
            
            if files:
                await cl.Message("Data received! Analyzing...").send()
                # Process uploaded data
                
    elif profile == "admin":
        # Handle admin commands
        if message.content.lower().startswith("/status"):
            await cl.Message("System Status: All services running normally").send()
        
    # Default assistant behavior for all profiles
    await cl.Message(f"[{profile}] Processing: {message.content}").send()

from typing import Optional, Dict, Any, List, Callable
from dataclasses import dataclass

# User and session types
class UserMetadata(Dict[str, Any]):
    """Type hint for user metadata dictionary"""
    pass

# Session accessor function type
SessionAccessor = Callable[[], Any]

# Thread dictionary structure for chat resume
ThreadDict = Dict[str, Any]  # Contains thread metadata and history