PySide6 QML Models & Services

The model and service layers form the backend of the MVC architecture. Models define data structures with Qt signal support; services handle all external interactions; the signal registry provides decoupled cross-layer communication.

Model Layer

BaseModel

All domain models inherit from BaseModel to gain Qt signal support and serialization:

"""models/base.py""" from datetime import datetime from typing import Any, TypeVar, Type from PySide6.QtCore import QObject, Signal

T = TypeVar('T', bound='BaseModel')

class BaseModel(QObject): """ Base class for all domain models.

Provides:
- property_changed(name, value) signal for reactive updates
- changed() signal for any modification
- _set_property() helper for consistent change notification
- to_dict() / from_dict() for serialization
"""

property_changed = Signal(str, object)  # name, value
changed = Signal()

def __init__(self, parent: QObject | None = None) -> None:
    super().__init__(parent)
    self._created_at: datetime = datetime.now()
    self._updated_at: datetime = datetime.now()

@property
def created_at(self) -> datetime:
    return self._created_at

@property
def updated_at(self) -> datetime:
    return self._updated_at

def _set_property(self, name: str, old_value: Any, new_value: Any) -> bool:
    """Set a property, emit signals if changed. Returns True if changed."""
    if old_value != new_value:
        self._updated_at = datetime.now()
        self.property_changed.emit(name, new_value)
        self.changed.emit()
        return True
    return False

def to_dict(self) -> dict[str, Any]:
    raise NotImplementedError("Subclasses must implement to_dict()")

@classmethod
def from_dict(cls: Type[T], data: dict[str, Any]) -> T:
    raise NotImplementedError("Subclasses must implement from_dict()")

Domain Model Example

"""models/job.py""" from dataclasses import dataclass, field from datetime import datetime from enum import Enum from pathlib import Path from typing import Any, Optional

from PySide6.QtCore import Signal from my_app.models.base import BaseModel

class JobStatus(Enum): ACTIVE = "active" COMPLETE = "complete" ARCHIVED = "archived" ON_HOLD = "on_hold"

@dataclass class JobFile: """Value object for files associated with a job.""" path: Path file_type: str file_name: str size_bytes: int = 0 modified_at: datetime = field(default_factory=datetime.now)

def to_dict(self) -> dict[str, Any]:
    return {
        "path": str(self.path),
        "file_type": self.file_type,
        "file_name": self.file_name,
        "size_bytes": self.size_bytes,
        "modified_at": self.modified_at.isoformat(),
    }

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "JobFile":
    return cls(
        path=Path(data["path"]),
        file_type=data["file_type"],
        file_name=data["file_name"],
        size_bytes=data.get("size_bytes", 0),
        modified_at=datetime.fromisoformat(data.get("modified_at", "")),
    )

class Job(BaseModel): """Domain model for a production job."""

files_updated = Signal()

def __init__(
    self,
    job_number: str,
    job_name: str = "",
    status: JobStatus = JobStatus.ACTIVE,
    parent=None,
) -> None:
    super().__init__(parent)
    self._job_number = job_number
    self._job_name = job_name
    self._status = status
    self._files: list[JobFile] = []

# --- Properties with change notification ---

@property
def job_number(self) -> str:
    return self._job_number

@property
def job_name(self) -> str:
    return self._job_name

@job_name.setter
def job_name(self, value: str) -> None:
    if self._set_property("job_name", self._job_name, value):
        self._job_name = value

@property
def status(self) -> JobStatus:
    return self._status

@status.setter
def status(self, value: JobStatus) -> None:
    if self._set_property("status", self._status, value):
        self._status = value

@property
def files(self) -> list[JobFile]:
    return list(self._files)

def add_file(self, file: JobFile) -> None:
    self._files.append(file)
    self.files_updated.emit()
    self.changed.emit()

# --- Serialization ---

def to_dict(self) -> dict[str, Any]:
    return {
        "job_number": self._job_number,
        "job_name": self._job_name,
        "status": self._status.value,
        "files": [f.to_dict() for f in self._files],
        "created_at": self._created_at.isoformat(),
        "updated_at": self._updated_at.isoformat(),
    }

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Job":
    job = cls(
        job_number=data["job_number"],
        job_name=data.get("job_name", ""),
        status=JobStatus(data.get("status", "active")),
    )
    for f in data.get("files", []):
        job._files.append(JobFile.from_dict(f))
    return job

Model Design Rules

Rule Rationale

Always use _set_property() for mutable fields Ensures signals fire consistently

Use @dataclass for value objects (no signals needed) Lightweight, immutable data

Use BaseModel for entities tracked by controllers Reactive updates to UI

Never import services or views in models Pure data layer

Implement to_dict() / from_dict()

Enables persistence and serialization

Use Enums for constrained values Type safety, IDE autocomplete

Application State

A singleton that tracks transient UI state (not persisted domain data):

"""models/state.py""" from PySide6.QtCore import QObject, Signal

class ApplicationState(QObject): """ Singleton for transient application state.

Tracks current UI state such as active selection,
navigation position, and loading flags.
"""

active_job_changed = Signal(str)  # job_id or ""
navigation_changed = Signal(int)  # page index

_instance = None

def __new__(cls):
    if cls._instance is None:
        cls._instance = super().__new__(cls)
        cls._instance._init_state()
    return cls._instance

def _init_state(self):
    super().__init__()
    self._active_job_id: str | None = None
    self._current_page: int = 0

@property
def active_job_id(self) -> str | None:
    return self._active_job_id

@active_job_id.setter
def active_job_id(self, value: str | None) -> None:
    if self._active_job_id != value:
        self._active_job_id = value
        self.active_job_changed.emit(value or "")

@property
def current_page(self) -> int:
    return self._current_page

@current_page.setter
def current_page(self, value: int) -> None:
    if self._current_page != value:
        self._current_page = value
        self.navigation_changed.emit(value)

Signal Registry

Central registry of all application-wide signals. Views, controllers, and bridges connect to these:

"""utils/signals.py""" from PySide6.QtCore import QObject, Signal

class SignalRegistry(QObject): """ Central signal definitions for the application.

All cross-layer communication flows through these signals.
Organized by domain area.
"""

# --- Job Signals ---
job_changed = Signal(str)           # job_id
job_created = Signal(str)           # job_id
job_updated = Signal(str, str)      # job_id, field
job_deleted = Signal(str)           # job_id
job_files_updated = Signal(str, int)  # job_id, file_count

# --- Settings Signals ---
settings_changed = Signal(str, object)  # key, value
settings_reset = Signal(str)            # category or ""

# --- Connection Signals ---
broker_status_changed = Signal(bool)     # connected
broker_message_received = Signal(str, str)  # topic, payload

# --- UI Signals ---
theme_changed = Signal(str)         # "light" or "dark"
view_focus_requested = Signal(str)  # view_name
error_occurred = Signal(str, str)   # title, message
app_closing = Signal()

Singleton accessor

_registry: SignalRegistry | None = None

def get_signal_registry() -> SignalRegistry: global _registry if _registry is None: _registry = SignalRegistry() return _registry

Service Layer

Repository Pattern

"""services/database/jobs_repository.py""" import sqlite3 import logging from pathlib import Path from typing import Optional

from my_app.models.job import Job

logger = logging.getLogger(name)

class JobsRepository: """ SQLite-backed repository for job persistence.

Follows the repository pattern — controllers call these methods,
never raw SQL.
"""

def __init__(self, db_path: Path) -> None:
    self._db_path = db_path
    self._ensure_schema()

def _ensure_schema(self) -> None:
    with sqlite3.connect(self._db_path) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS jobs (
                job_number TEXT PRIMARY KEY,
                job_name TEXT DEFAULT '',
                status TEXT DEFAULT 'active',
                data_json TEXT DEFAULT '{}',
                created_at TEXT,
                updated_at TEXT
            )
        """)

def get_job(self, job_number: str) -> Optional[Job]:
    with sqlite3.connect(self._db_path) as conn:
        conn.row_factory = sqlite3.Row
        row = conn.execute(
            "SELECT * FROM jobs WHERE job_number = ?",
            (job_number,),
        ).fetchone()
    if row:
        return self._row_to_job(row)
    return None

def save_job(self, job: Job) -> None:
    import json
    data = job.to_dict()
    with sqlite3.connect(self._db_path) as conn:
        conn.execute("""
            INSERT OR REPLACE INTO jobs
            (job_number, job_name, status, data_json, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?)
        """, (
            data["job_number"],
            data["job_name"],
            data["status"],
            json.dumps(data),
            data["created_at"],
            data["updated_at"],
        ))

def get_all_jobs(self) -> list[Job]:
    with sqlite3.connect(self._db_path) as conn:
        conn.row_factory = sqlite3.Row
        rows = conn.execute(
            "SELECT * FROM jobs ORDER BY updated_at DESC"
        ).fetchall()
    return [self._row_to_job(r) for r in rows]

def delete_job(self, job_number: str) -> bool:
    with sqlite3.connect(self._db_path) as conn:
        cursor = conn.execute(
            "DELETE FROM jobs WHERE job_number = ?",
            (job_number,),
        )
    return cursor.rowcount > 0

def _row_to_job(self, row) -> Job:
    import json
    data = json.loads(row["data_json"])
    return Job.from_dict(data)

Service Pattern

Services wrap external interactions — file I/O, network calls, subprocess execution:

"""services/file_discovery.py""" import logging from pathlib import Path

logger = logging.getLogger(name)

class FileDiscoveryService: """Discovers and indexes files in job directories."""

SUPPORTED_EXTENSIONS = {".dxf", ".pdf", ".top", ".xml", ".nc"}

def __init__(self, base_path: Path) -> None:
    self._base_path = base_path

def initialize(self) -> None:
    self._base_path.mkdir(parents=True, exist_ok=True)

def shutdown(self) -> None:
    pass

def discover_files(self, job_number: str) -> list[dict]:
    job_dir = self._base_path / job_number
    if not job_dir.exists():
        return []

    results = []
    for path in job_dir.rglob("*"):
        if path.is_file() and path.suffix.lower() in self.SUPPORTED_EXTENSIONS:
            results.append({
                "path": str(path),
                "file_type": path.suffix.lstrip(".").lower(),
                "file_name": path.name,
                "size_bytes": path.stat().st_size,
            })
    return results

Service Design Rules

Rule Rationale

Services have initialize() and shutdown() lifecycle methods Clean startup/teardown

Services never import models, views, or controllers Pure I/O layer

Services return plain data (dicts, lists, primitives) or model instances No Qt types leaked

One service per external system Database, file I/O, broker, API each get their own

Services are injected via constructor, never imported directly by views Testable, replaceable

Controller Layer

Controllers wire models and services together, emitting signals for the bridge/view layer:

"""controllers/base.py""" from typing import Any, Protocol from PySide6.QtCore import QObject from my_app.utils.signals import SignalRegistry

class IView(Protocol): def bind(self, controller: "BaseController") -> None: ... def update_display(self, data: dict[str, Any]) -> None: ...

class BaseController(QObject): """ Base class for all controllers.

Provides signal registry access, view registration,
and lifecycle management.
"""

def __init__(self, signals: SignalRegistry, parent: QObject | None = None):
    super().__init__(parent)
    self._signals = signals
    self._views: list[IView] = []
    self._initialized = False

@property
def signals(self) -> SignalRegistry:
    return self._signals

def register_view(self, view: IView) -> None:
    if view not in self._views:
        self._views.append(view)
        view.bind(self)

def notify_views(self, data: dict[str, Any] | None = None) -> None:
    for view in self._views:
        view.update_display(data or {})

def initialize(self) -> None:
    self._initialized = True

def cleanup(self) -> None:
    self._views.clear()

Domain Controller Example

"""controllers/job_controller.py""" import logging from PySide6.QtCore import QObject

from my_app.controllers.base import BaseController from my_app.models.job import Job, JobStatus from my_app.models.state import ApplicationState from my_app.services.database.jobs_repository import JobsRepository from my_app.utils.signals import SignalRegistry

logger = logging.getLogger(name)

class JobController(BaseController): """ Handles job operations: activate, create, update, search.

Delegates persistence to JobsRepository and
emits signals via SignalRegistry for UI updates.
"""

def __init__(
    self,
    signals: SignalRegistry,
    repository: JobsRepository,
    parent: QObject | None = None,
) -> None:
    super().__init__(signals, parent)
    self._repository = repository
    self._state = ApplicationState()

def initialize(self) -> None:
    super().initialize()
    if self._state.active_job_id:
        job = self._repository.get_job(self._state.active_job_id)
        if job is None:
            self._state.active_job_id = None

def activate_job(self, job_id: str) -> bool:
    job = self._repository.get_job(job_id)
    if job is None:
        logger.warning(f"Job {job_id} not found")
        return False

    self._state.active_job_id = job_id
    self._signals.job_changed.emit(job_id)
    logger.info(f"Activated job {job_id}")
    return True

def create_job(self, job_number: str) -> bool:
    if self._repository.get_job(job_number):
        logger.warning(f"Job {job_number} already exists")
        return False

    job = Job(job_number=job_number)
    self._repository.save_job(job)
    self._signals.job_created.emit(job_number)
    return True

def get_job(self, job_id: str) -> Job | None:
    return self._repository.get_job(job_id)

def get_all_jobs(self) -> list[Job]:
    return self._repository.get_all_jobs()

Controller Design Rules

Rule Rationale

Controllers receive services via constructor injection Testable, no hidden dependencies

Controllers emit signals, never manipulate UI Decoupled from view layer

One controller per domain area Job, Settings, Script, etc.

Controllers read/write via repositories, not raw DB Abstraction, testability

Controllers update ApplicationState for UI-relevant state Centralized state tracking

Testing Approach

"""tests/test_job_controller.py""" from unittest.mock import MagicMock from my_app.controllers.job_controller import JobController from my_app.utils.signals import SignalRegistry from my_app.models.job import Job

def test_activate_job_emits_signal(): signals = SignalRegistry() repo = MagicMock() repo.get_job.return_value = Job(job_number="1234567")

controller = JobController(signals=signals, repository=repo)
handler = MagicMock()
signals.job_changed.connect(handler)

result = controller.activate_job("1234567")

assert result is True
handler.assert_called_once_with("1234567")

def test_activate_nonexistent_job_returns_false(): signals = SignalRegistry() repo = MagicMock() repo.get_job.return_value = None

controller = JobController(signals=signals, repository=repo)
result = controller.activate_job("9999999")

assert result is False

References

• PySide6 Signals and Slots

• PySide6 QAbstractListModel

• Repository Pattern