Skip to content

01 application context

Jump to bottom
Zuko edited this page Jan 21, 2026 · 1 revision

QtAppContext - Application Lifecycle Manager

Central orchestrator cho application context, services, và lifecycle

Overview

QtAppContext là singleton quản lý toàn bộ application lifecycle, từ bootstrap đến shutdown. Nó cung cấp:

  • Feature flags (enable/disable components via environment variables)
  • Service registry (global services)
  • Scoped resource management (task-specific cleanup)
  • Application state management
  • Qt event loop integration

API Reference

Singleton Access

from core import QtAppContext

ctx = QtAppContext.globalInstance()  # Thread-safe singleton

Bootstrap

ctx.bootstrap()  # Idempotent - chỉ chạy 1 lần

Bootstrap process:

  1. Load .env file (nếu có python-dotenv)
  2. Setup async event loop (qasync)
  3. Initialize Config singleton
  4. Initialize Publisher singleton
  5. Setup global exception handler
  6. Conditionally initialize NetworkManager (nếu enabled)
  7. Conditionally initialize TaskManagerService (nếu enabled)
  8. Emit app.ready event

Feature Flags

# Check if feature enabled
if ctx.isFeatureEnabled('network'):
    network = ctx.network

if ctx.isFeatureEnabled('tasks'):
    taskManager = ctx.taskManager

Environment variables:

# .env file
PSA_ENABLE_NETWORK=true   # Default: true
PSA_ENABLE_TASKS=true     # Default: true

Parsing rules:

  • true, 1, yes, on → True
  • false, 0, no, off → False
  • Missing → Default value

Core Services

# Always available
config = ctx.config          # Config singleton
publisher = ctx.publisher    # Publisher singleton

# Conditional (check feature flags first)
network = ctx.network        # QNetworkAccessManager | None
taskManager = ctx.taskManager  # TaskManagerService | None

Service Registration

Global services:

# Register
myService = MyService()
ctx.registerService('my_service', myService)

# Retrieve
myService = ctx.getService('my_service')

Scoped services:

# Register with tag (usually task UUID)
taskId = str(uuid.uuid4())
browser = ChromeBrowserService()
ctx.registerScopedService(taskId, browser)

# Cleanup all services under tag
ctx.releaseScope(taskId)  # Calls cleanup()/close()/dispose()

State Management

# Set shared state
ctx.setState('current_user', {'id': 123, 'name': 'John'})

# Get shared state
user = ctx.getState('current_user')
user = ctx.getState('missing_key', default={'id': 0})

Thread-safe: Uses QMutex internally.

Lifecycle Signals

# Connect to lifecycle signals
ctx.appBooting.connect(onAppBooting)
ctx.appReady.connect(onAppReady)
ctx.appClosing.connect(onAppClosing)

Run Event Loop

exitCode = ctx.run()  # Blocks until app quits

Note: Automatically calls bootstrap() if not already done.

Usage Examples

Basic Application

from core import QtAppContext
from app.windows.main import MainWindow

def main():
    # 1. Get context
    ctx = QtAppContext.globalInstance()
    
    # 2. Bootstrap
    ctx.bootstrap()
    
    # 3. Create main window
    mainWindow = MainWindow()
    mainWindow.show()
    
    # 4. Run event loop
    return ctx.run()

if __name__ == '__main__':
    import sys
    sys.exit(main())

With Custom Services

from core import QtAppContext

class DatabaseService:
    def __init__(self, config):
        self.config = config
        self.connection = None
    
    def connect(self):
        # Connect to database
        pass

def main():
    ctx = QtAppContext.globalInstance()
    ctx.bootstrap()
    
    # Register custom service
    dbService = DatabaseService(ctx.config)
    ctx.registerService('database', dbService)
    dbService.connect()
    
    # Access from anywhere
    db = ctx.getService('database')
    
    return ctx.run()

Scoped Resources in Task

from core import QtAppContext
from core.taskSystem import AbstractTask

class BrowserAutomationTask(AbstractTask):
    def handle(self):
        ctx = QtAppContext.globalInstance()
        taskId = self.uuid
        
        # Create scoped resources
        browser = ChromeBrowserService()
        tempFiles = TempFileHandler()
        
        # Register for auto cleanup
        ctx.registerScopedService(taskId, browser)
        ctx.registerScopedService(taskId, tempFiles)
        
        try:
            # Use resources
            browser.navigate('https://example.com')
            tempFiles.createTemp('data.json')
            
            # Do work...
            if self.isStopped():
                return
            
        finally:
            # Auto cleanup: calls browser.cleanup() and tempFiles.cleanup()
            ctx.releaseScope(taskId)

Feature Flag Usage

from core import QtAppContext

ctx = QtAppContext.globalInstance()
ctx.bootstrap()

# Check before using
if ctx.isFeatureEnabled('network'):
    from PySide6.QtNetwork import QNetworkRequest
    from PySide6.QtCore import QUrl
    
    request = QNetworkRequest(QUrl('https://api.example.com'))
    reply = ctx.network.get(request)
else:
    # Fallback: use requests library
    import requests
    response = requests.get('https://api.example.com')

Lifecycle Hooks

from core import QtAppContext
from core.Logging import logger

def onAppBooting():
    logger.info('Application is booting...')

def onAppReady():
    logger.info('Application is ready!')
    # Initialize UI, load data, etc.

def onAppClosing():
    logger.info('Application is closing...')
    # Save state, cleanup resources

ctx = QtAppContext.globalInstance()
ctx.appBooting.connect(onAppBooting)
ctx.appReady.connect(onAppReady)
ctx.appClosing.connect(onAppClosing)

ctx.bootstrap()
ctx.run()

Architecture

graph TB
    App[Application] -->|globalInstance| QtAppContext
    QtAppContext -->|bootstrap| LoadEnv[Load .env]
    LoadEnv --> SetupAsync[Setup qasync]
    SetupAsync --> InitConfig[Init Config]
    InitConfig --> InitPublisher[Init Publisher]
    InitPublisher --> ExceptionHandler[Setup Exception Handler]
    ExceptionHandler --> CheckNetwork{Network Enabled?}
    CheckNetwork -->|Yes| InitNetwork[Init NetworkManager]
    CheckNetwork -->|No| CheckTasks
    InitNetwork --> CheckTasks{Tasks Enabled?}
    CheckTasks -->|Yes| InitTasks[Init TaskManagerService]
    CheckTasks -->|No| EmitReady
    InitTasks --> EmitReady[Emit app.ready]
    
    QtAppContext --> ServiceLocator
    ServiceLocator --> GlobalServices[Global Services]
    ServiceLocator --> ScopedServices[Scoped Services]
    
    QtAppContext --> SharedState[Shared State Dict]
    
    style QtAppContext fill:#e1f5ff
    style ServiceLocator fill:#fff4e1
    style EmitReady fill:#c8e6c9
Loading

Best Practices

✅ DO

# Use singleton instance
ctx = QtAppContext.globalInstance()

# Bootstrap before accessing services
ctx.bootstrap()

# Check feature flags before using optional services
if ctx.isFeatureEnabled('network'):
    network = ctx.network

# Use scoped services for task-specific resources
ctx.registerScopedService(taskId, resource)
try:
    # Use resource
    pass
finally:
    ctx.releaseScope(taskId)

# Register global services during bootstrap
ctx.bootstrap()
myService = MyService()
ctx.registerService('my_service', myService)

❌ DON'T

# Don't create multiple instances
ctx1 = QtAppContext()  # Wrong! Use globalInstance()

# Don't access services before bootstrap
ctx = QtAppContext.globalInstance()
config = ctx.config  # Wrong! Bootstrap first

# Don't use NetworkManager in background threads
def background_task():
    ctx = QtAppContext.globalInstance()
    network = ctx.network  # Wrong! UI thread only
    # Use requests library instead

# Don't forget to release scoped services
ctx.registerScopedService(taskId, browser)
# ... use browser ...
# Missing: ctx.releaseScope(taskId)  # Memory leak!

# Don't register services before bootstrap
ctx = QtAppContext.globalInstance()
ctx.registerService('my_service', MyService())  # Wrong order
ctx.bootstrap()

Thread Safety

  • ✅ globalInstance(): Thread-safe (QMutex)
  • ✅ bootstrap(): Thread-safe, idempotent
  • ✅ setState()/getState(): Thread-safe (QMutex)
  • ✅ Service registration: Thread-safe (delegated to ServiceLocator)
  • ⚠️ network: UI thread only (QNetworkAccessManager limitation)

Common Patterns

Application Entry Point

# main.py
from core import QtAppContext
from app.windows.main import MainWindow

def main():
    ctx = QtAppContext.globalInstance()
    ctx.bootstrap()
    
    mainWindow = MainWindow()
    mainWindow.show()
    
    return ctx.run()

if __name__ == '__main__':
    import sys
    sys.exit(main())

Service Access Pattern

# Anywhere in application
from core import QtAppContext

ctx = QtAppContext.globalInstance()
config = ctx.config
publisher = ctx.publisher
myService = ctx.getService('my_service')

Task Scoping Pattern

class MyTask(AbstractTask):
    def handle(self):
        ctx = QtAppContext.globalInstance()
        taskId = self.uuid
        
        # Setup scoped resources
        resources = [
            ChromeBrowserService(),
            TempFileHandler(),
            ApiSession()
        ]
        
        for resource in resources:
            ctx.registerScopedService(taskId, resource)
        
        try:
            # Task logic
            pass
        finally:
            ctx.releaseScope(taskId)

Related Documentation

Troubleshooting

Q: Services are None after bootstrap

# Check feature flags
ctx = QtAppContext.globalInstance()
ctx.bootstrap()

if ctx.network is None:
    # Check PSA_ENABLE_NETWORK in .env
    print(ctx.isFeatureEnabled('network'))

Q: Bootstrap called multiple times

# Safe - idempotent
ctx.bootstrap()
ctx.bootstrap()  # Logs warning, does nothing

Q: Scoped services not cleaned up

# Ensure cleanup() method exists
class MyService:
    def cleanup(self):  # Priority 1
        # Cleanup logic
        pass
    
    def close(self):  # Priority 2 (if cleanup missing)
        pass
    
    def dispose(self):  # Priority 3 (if both missing)
        pass

Clone this wiki locally