Simpro API Client Foundation

Build a production-grade Python OAuth2 HTTP client for the Simpro API.

Introduction

30 Second Summary

Every business system holds valuable data behind an API. Connecting to that API means solving three problems that trip up most developers: authenticating securely, handling token expiry gracefully, and managing HTTP errors without crashing.

In this project, you will build a production-grade Python package that handles OAuth2 authentication and HTTP transport for the Simpro API. The package obtains tokens, caches them, refreshes them automatically, and gives you a clean client you can import anywhere.

What You'll Build

By the end of this sprint, you can import your package, point it at any Simpro-compatible API endpoint, and make authenticated requests that handle token refresh, rate limits, and errors without any manual intervention.

By the end of this project, you'll have:

  • A configuration module that validates every setting on startup and tells you exactly what's wrong if something is missing.
  • An auth manager that obtains, caches, and refreshes OAuth2 tokens automatically so downstream code never worries about authentication.
  • A base HTTP client with connection pooling, timeouts, and typed exceptions that you can call and immediately see structured error responses when something goes wrong.
  • A full test suite proving every scenario works without making a single real API call.
  • Secret Mission: Add structured logging with correlation IDs that trace every request through the system.

Are there any prerequisites?

This sprint builds on an existing CLIVE platform with Phase 1 and Phase 2 complete. You need Ubuntu 24.04 LTS with Docker, PostgreSQL, and Python 3.12.3 already installed.

Before We Start

Before diving into code, let's lock in what you're about to build and why it matters. The CLIVE platform needs to pull live data from the Simpro API, and a dedicated integration layer keeps authentication, error handling, and configuration in one place rather than scattered across ad-hoc calls.

Set Up the Project Scaffold

Without a proper project structure and pinned dependencies, Python code becomes unreproducible. Libraries update silently, and what works today breaks tomorrow when a transient version mismatch creeps in.

In this step, you will create the phase3 project directory, declare every dependency with an exact version pin, and confirm the scaffold installs and lints cleanly.

In this step, get ready to:
  • Create the phase3 directory structure with a src/simpro_client/ package layout.
  • Define all dependencies with exact version pins in pyproject.toml.
  • Install everything in a virtual environment and confirm linting passes.
Create the project directory structure

A Python package using src layout keeps importable code separate from project metadata, tests, and configuration files. This prevents accidental imports from the project root during development.

  • Create the full directory tree by running these commands:
cd ~
mkdir -p phase3/src/simpro_client phase3/tests
cd phase3
  • Confirm the directories exist by running:
find . -type d

You should see ./src/simpro_client and ./tests in the output.

Not seeing the directories?

Make sure you ran the mkdir -p command from your home directory. The -p flag creates all parent directories in one go.

help me debug my directory structure for the phase3 project

Define the project metadata and dependencies

The pyproject.toml file is the single source of truth for your package's identity, dependencies, and tool configuration. Pinning exact versions ensures every install produces the same environment.

  • Create a new file called pyproject.toml in the phase3 directory and add the project metadata and dependencies:
[build-system]
requires = ["setuptools>=75.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "simpro-client"
version = "0.1.0"
description = "Reusable integration layer for the Simpro REST API"
requires-python = ">=3.12"
license = {text = "MIT"}
dependencies = [
    "httpx==0.28.1",
    "pydantic==2.13.4",
    "pydantic-settings==2.14.2",
    "python-dotenv==1.1.0",
]

[project.optional-dependencies]
dev = [
    "pytest==9.1.1",
    "respx==0.23.1",
    "ruff==0.16.1",
    "pytest-cov==7.0.0",
]

[tool.setuptools.packages.find]
where = ["src"]

What does this define?

  • build-system tells pip how to build the package (using setuptools).
  • dependencies pins the four runtime libraries: httpx for HTTP requests, pydantic for data validation, pydantic-settings for config loading, and python-dotenv for .env file support.
  • dev optional dependencies add testing and linting tools that are not needed at runtime.
  • packages.find where = ["src"] tells setuptools to look for importable packages inside the src/ directory.
  • Below the [tool.setuptools.packages.find] section, add the tool configuration:
[tool.ruff]
target-version = "py312"
line-length = 88

[tool.ruff.lint]
select = ["E", "F", "I", "N", "W", "UP", "B", "A", "SIM"]

[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]

What do these tool settings do?

  • tool.ruff configures the linter to target Python 3.12 and enforce an 88-character line width.
  • The select list enables rule categories for errors, pyflakes, import sorting, naming, warnings, upgrade suggestions, bugbear, builtins shadowing, and code simplification.
  • tool.pytest tells pytest to find tests in tests/ and add src/ to the Python path so imports resolve correctly.
  • Save pyproject.toml.

✔️ Awesome, I've got everything!

Great. Make sure you have saved pyproject.toml with both sections (project metadata and tool configuration) before moving on.

ⓧ I'd like to double check the full code

Your pyproject.toml should contain both chunks from above, in order:

  • The first section starting with [build-system] through [tool.setuptools.packages.find] (project metadata, runtime dependencies, dev dependencies, and package discovery).
  • The second section starting with [tool.ruff] through [tool.pytest.ini_options] (linter and test runner configuration).

Scroll up and confirm each block is present with no extra blank lines between sections.

  • Now create the package init file at src/simpro_client/__init__.py with the following content:
"""Simpro API Client - Reusable integration layer for the Simpro REST API."""

from simpro_client.client import SimproClient
from simpro_client.config import SimproSettings, get_settings

__all__ = ["SimproClient", "SimproSettings", "get_settings"]
__version__ = "0.1.0"

What does this file do?

The __init__.py file marks simpro_client as an importable Python package. It pre-imports the main classes so users can write from simpro_client import SimproClient directly.

The imports reference modules (client and config) that you will create in later steps. Ruff does not verify whether imported modules exist on disk, so the linter will pass without them.

  • Save src/simpro_client/__init__.py.

✔️ Awesome, I've got everything!

Nice. You now have the two core files that define your package.

ⓧ I'd like to double check the full code

"""Simpro API Client - Reusable integration layer for the Simpro REST API."""

from simpro_client.client import SimproClient
from simpro_client.config import SimproSettings, get_settings

__all__ = ["SimproClient", "SimproSettings", "get_settings"]
__version__ = "0.1.0"
Set up the virtual environment and install dependencies

A virtual environment isolates this project's packages from the system Python. Installing in editable mode (-e) means code changes take effect immediately without reinstalling.

  • Create the virtual environment by running:
python3 -m venv .venv
  • Activate it by running:
source .venv/bin/activate

Your terminal prompt should now show (.venv) at the beginning.

  • Install the package and all dev dependencies in editable mode by running:
pip install -e ".[dev]"

Pip will download and install all pinned packages. You should see a stream of install progress ending with Successfully installed followed by the package list.

Seeing a build error?

  • Confirm you are inside the phase3 directory (run pwd to check). The pyproject.toml must be in your current directory.
  • Check that your virtual environment is active. You should see (.venv) in your prompt.
  • If you see a TOML parse error, double-check the bracket sections and quotes in pyproject.toml match exactly.

help me fix a pip install error for my simpro-client package

Confirm linting passes

The final check proves the scaffold is valid. If ruff finds zero issues, your project structure and code style are correct.

  • Before you run this, what do you expect ruff to report for a package with only an __init__.py that imports from modules that do not exist yet?
  • Run the linter by running:
ruff check src/

You should see the command return with no output and no error. Ruff performs static analysis only, so it does not check whether imported modules exist on disk.

  • Also confirm formatting is clean by running:
ruff format --check src/

You should see 1 file already formatted confirming no formatting changes are needed.

Seeing ruff warnings?

  • If ruff reports an F401 (unused import), make sure your __all__ list references every imported name. Ruff treats names in __all__ as used.
  • If you get command not found: ruff, make sure your virtual environment is activated (look for (.venv) in your prompt).

help me fix ruff linting errors in my simpro_client package

Your project scaffold is locked in. Every dependency is pinned, the package structure is valid, and the linter is happy. Next up, you will build the configuration module that loads and validates all Simpro connection settings from environment variables.

Build the Configuration Module

In the previous step you stood up the project scaffold with all dependencies installed and linting passing. Now it is time to write the module that every other part of the client depends on: configuration.

Hardcoded credentials and URLs cause security incidents and make code impossible to reuse across environments. A validated configuration module catches mistakes at startup rather than letting them surface as confusing runtime errors deep inside an API call.

In this step, get ready to:
  • Create a configuration class that validates all Simpro connection settings on startup.
  • Document every environment variable and protect secrets from version control.
  • Write tests proving the module loads correctly and rejects missing fields.
Create the settings class

The configuration module uses pydantic-settings to declare every setting the Simpro client needs. Each field is validated at instantiation time, so a typo or missing credential fails fast with a clear message.

  • Create a new file at src/simpro_client/config.py and add the class skeleton with the connection URL fields by pasting this code:
"""Configuration management using pydantic-settings."""

from functools import lru_cache

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class SimproSettings(BaseSettings):
    """Simpro API connection settings.

    All values are loaded from environment variables or a .env file.
    """

    model_config = SettingsConfigDict(
        env_file=".env",
        env_prefix="SIMPRO_",
        case_sensitive=False,
    )

    base_url: str = Field(
        description="Base URL of the Simpro API (e.g. http://simpro-mock:8000/api/v1.0)"
    )
    token_url: str = Field(
        description="OAuth2 token endpoint URL"
    )
    client_id: str = Field(
        description="OAuth2 Client ID"
    )

What does this code do?

SettingsConfigDict tells pydantic-settings to look for a .env file and to strip the SIMPRO_ prefix from environment variable names when matching them to fields.

The three fields declared so far (base_url, token_url, client_id) have no default values. That makes them required. If any is missing at startup, Pydantic raises a validation error immediately.

  • Add the remaining fields directly below client_id (still inside the class body) by pasting this code:
    client_secret: str = Field(
        description="OAuth2 Client Secret"
    )
    api_key: str | None = Field(
        default=None,
        description="Static API key (fallback auth, optional)",
    )
    auth_mode: str = Field(
        default="client_credentials",
        description="Auth mode: 'client_credentials' or 'api_key'",
    )
    company_id_service: int = Field(
        default=1,
        description="Company ID for CVC Service",
    )
    company_id_projects: int = Field(
        default=2,
        description="Company ID for CVC Projects",
    )
    timeout: float = Field(
        default=30.0,
        description="HTTP request timeout in seconds",
    )
    max_retries: int = Field(
        default=3,
        description="Maximum number of retries on transient failures",
    )

What do these fields control?

client_secret is the fourth required field. Together with client_id, it forms the OAuth2 Client Credentials pair.

The remaining fields all carry defaults, making them optional in the environment. auth_mode lets you switch between OAuth2 and a static API key. company_id_service and company_id_projects map to the CVC multi-company setup in Simpro.

  • Save config.py and verify the class imports cleanly by running:
python -c "from simpro_client.config import SimproSettings; print('SimproSettings loaded')"

You should see SimproSettings loaded printed with no errors.

Seeing an ImportError?

Make sure your virtual environment is activated (source .venv/bin/activate). If you see an error about simpro_client.client, that is expected. The __init__.py imports a module that does not exist yet. Importing directly from simpro_client.config avoids that.

help me debug this ImportError in my config module

  • Now add the factory function below the class (outside the class body, at the module level) by pasting this code at the bottom of config.py:


@lru_cache
def get_settings() -> SimproSettings:
    """Get cached application settings."""
    return SimproSettings()

Why use lru_cache here?

get_settings() reads environment variables and validates them every time it is called. Wrapping it with @lru_cache ensures this work happens only once. Every subsequent call returns the same validated instance instantly.

  • Save config.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm your config.py contains these sections in order:

  • Module docstring and imports (lru_cache, Field, BaseSettings, SettingsConfigDict).
  • SimproSettings class with model_config and 10 fields (base_url through max_retries).
  • get_settings() function decorated with @lru_cache at the bottom.
Document and protect environment variables

A .env.example file documents every variable the module expects, so anyone cloning the repo knows what to configure. The actual .env with real credentials stays out of version control.

  • Create a file called .env.example in your phase3/ directory (the same level as pyproject.toml) and paste this content:
# Simpro API Configuration
# Copy this file to .env and fill in your values

# Base URL of the Simpro API (change to live URL when real access is enabled)
SIMPRO_BASE_URL=http://simpro-mock:8000/api/v1.0

# OAuth2 token endpoint
SIMPRO_TOKEN_URL=http://simpro-mock:8000/oauth2/token

# OAuth2 Client Credentials
SIMPRO_CLIENT_ID=your-client-id-here
SIMPRO_CLIENT_SECRET=your-client-secret-here

# Auth mode: 'client_credentials' or 'api_key'
SIMPRO_AUTH_MODE=client_credentials

# Static API key (only used when SIMPRO_AUTH_MODE=api_key)
# SIMPRO_API_KEY=your-api-key-here

# Company IDs for CVC multi-company setup
SIMPRO_COMPANY_ID_SERVICE=1
SIMPRO_COMPANY_ID_PROJECTS=2

# HTTP settings
SIMPRO_TIMEOUT=30.0
SIMPRO_MAX_RETRIES=3

What does each variable control?

SIMPRO_BASE_URL and SIMPRO_TOKEN_URL point to the mock server you will build in Sprint 2. SIMPRO_CLIENT_ID and SIMPRO_CLIENT_SECRET are the OAuth2 credentials.

The SIMPRO_API_KEY line is commented out because it is only needed in API key mode. Company IDs, timeout, and max retries all have sensible defaults.

  • Add .env to your .gitignore so real credentials never get committed by running:
echo ".env" >> .gitignore
  • Copy the example to a real .env file so you can test the module by running:
cp .env.example .env

Before you run the next command, what do you think get_settings() will print for base_url?

  • Test that settings load from the .env file by running:
python -c "from simpro_client.config import get_settings; print(get_settings().base_url)"

You should see http://simpro-mock:8000/api/v1.0 printed. The module read the value straight from your .env file and validated it.

Seeing a ValidationError instead?

Check that your .env file is in the same directory you are running the command from (the phase3/ root). Also confirm the file contains all four required variables: SIMPRO_BASE_URL, SIMPRO_TOKEN_URL, SIMPRO_CLIENT_ID, SIMPRO_CLIENT_SECRET.

help me fix a ValidationError when loading my Simpro settings

Write and run configuration tests

Automated tests lock in the behaviour you just verified manually. They also prove the module handles bad input gracefully, which is hard to test interactively.

  • Create the tests/ directory and its __init__.py by running:
mkdir -p tests && echo '"""Test suite for simpro_client."""' > tests/__init__.py
  • Confirm the directory exists by running:
ls tests/

You should see __init__.py listed.

  • Create tests/conftest.py with the shared test fixtures by pasting this code:
"""Shared test fixtures."""

import pytest

from simpro_client.config import SimproSettings


@pytest.fixture
def mock_settings() -> SimproSettings:
    """Create test settings without requiring .env file."""
    return SimproSettings(
        base_url="http://test.simprosuite.com/api/v1.0",
        token_url="http://test.simprosuite.com/oauth2/token",
        client_id="test-client-id",
        client_secret="test-client-secret",
        auth_mode="client_credentials",
        company_id_service=1,
        company_id_projects=2,
        timeout=5.0,
        max_retries=3,
    )

Why test fixtures?

Fixtures create reusable test data. The mock_settings fixture builds a SimproSettings instance with hardcoded values so tests never depend on your real .env file. Every test that needs settings can request this fixture by name.

  • Add a second fixture for API key mode below the first one in the same file:


@pytest.fixture
def api_key_settings() -> SimproSettings:
    """Create test settings for API key mode."""
    return SimproSettings(
        base_url="http://test.simprosuite.com/api/v1.0",
        token_url="http://test.simprosuite.com/oauth2/token",
        client_id="unused",
        client_secret="unused",
        auth_mode="api_key",
        api_key="test-static-token",
        company_id_service=1,
        company_id_projects=2,
        timeout=5.0,
        max_retries=3,
    )
  • Save conftest.py and verify pytest can see the fixtures by running:
pytest --collect-only

You should see no tests ran or a message that 0 items were collected. No errors means the fixtures loaded cleanly.

  • Create tests/test_config.py and add the imports plus the first test function:
"""Tests for configuration module."""

import pytest
from pydantic import ValidationError

from simpro_client.config import SimproSettings


def test_settings_load_from_values():
    """Settings load correctly when all values are provided."""
    settings = SimproSettings(
        base_url="http://localhost:8000/api/v1.0",
        token_url="http://localhost:8000/oauth2/token",
        client_id="my-id",
        client_secret="my-secret",
    )
    assert settings.base_url == "http://localhost:8000/api/v1.0"
    assert settings.client_id == "my-id"
    assert settings.timeout == 30.0
    assert settings.company_id_service == 1
    assert settings.company_id_projects == 2

What does this test prove?

It constructs a SimproSettings with only the four required fields and checks that defaults fill in the rest. If timeout or company_id_service come back wrong, you know the default wiring is broken.

  • Add the remaining two test functions below the first one in test_config.py:


def test_settings_missing_required_field():
    """Missing required fields raise a validation error."""
    with pytest.raises(ValidationError) as exc_info:
        SimproSettings(
            base_url="http://localhost:8000/api/v1.0",
            token_url="http://localhost:8000/oauth2/token",
            client_id="my-id",
            # client_secret is missing
        )
    assert "client_secret" in str(exc_info.value)


def test_settings_defaults():
    """Default values are applied correctly."""
    settings = SimproSettings(
        base_url="http://localhost:8000/api/v1.0",
        token_url="http://localhost:8000/oauth2/token",
        client_id="id",
        client_secret="secret",
    )
    assert settings.auth_mode == "client_credentials"
    assert settings.api_key is None
    assert settings.max_retries == 3

What do these tests cover?

test_settings_missing_required_field proves that omitting client_secret raises a ValidationError that names the missing field. This is the safety net that catches misconfigured deployments immediately.

test_settings_defaults confirms that optional fields (auth_mode, api_key, max_retries) carry the expected default values when not explicitly set.

  • Save test_config.py.

Before you run the tests, do you expect all three to pass on the first try?

  • Run the full configuration test suite by running:
pytest tests/test_config.py -v

You should see 3 passed with green output. Each test name shows as PASSED next to it.

Tests failing or not collected?

If pytest reports ModuleNotFoundError: No module named 'simpro_client', confirm you installed the package in editable mode (pip install -e ".[dev]" from the phase3/ directory).

If you see collection errors mentioning conftest.py, check that the file is inside the tests/ directory and not at the project root.

help me debug why pytest cannot find my simpro_client module or fixtures

Your configuration module is locked in with tests proving it works and fails gracefully. Next up, you will build the authentication manager that uses these settings to obtain and cache OAuth2 tokens automatically.

Build the Authentication Module

Your configuration module validates settings at startup, but settings alone cannot talk to an API. The Simpro API uses OAuth2 Client Credentials for authentication, which means your code must obtain a token, cache it, and refresh it before every request.

Getting token lifecycle management wrong is the most common source of confusing 401 errors in API integrations. Tokens expire silently, refresh flows fail without warning, and stale auth headers produce cryptic downstream failures. Building this once means every future endpoint module inherits correct auth automatically.

In this step, get ready to:
  • Create a typed exception hierarchy for all Simpro API errors.
  • Build an AuthManager class that handles OAuth2 token lifecycle and API key fallback.
  • Write tests proving token caching, refresh, and error handling all work without any external API calls.
Create the exception hierarchy

Before writing the auth module, you need a set of typed exceptions it can raise. A typed hierarchy lets downstream code catch specific failures (auth errors vs. rate limits vs. not-found) instead of parsing generic error messages.

  • Create a new file src/simpro_client/exceptions.py and add the base exception classes:
"""Typed exception hierarchy for Simpro API errors."""


class SimproError(Exception):
    """Base exception for all Simpro client errors."""

    def __init__(self, message: str) -> None:
        self.message = message
        super().__init__(message)


class SimproAuthError(SimproError):
    """Authentication or authorization failure."""

    pass


class SimproAPIError(SimproError):
    """API returned an error response."""

    def __init__(
        self, message: str, status_code: int, response_body: str = ""
    ) -> None:
        self.status_code = status_code
        self.response_body = response_body
        super().__init__(message)

What does this code do?

SimproError is the base that all Simpro exceptions inherit from. It stores a message attribute so callers can always access a human-readable error string.

SimproAuthError covers authentication failures (bad credentials, expired tokens). SimproAPIError captures any non-auth API error, carrying the HTTP status_code and response_body for debugging.

  • Below the SimproAPIError class, add the two specific subclasses:
class SimproRateLimitError(SimproAPIError):
    """Rate limit exceeded (HTTP 429)."""

    def __init__(
        self,
        message: str = "Rate limit exceeded",
        retry_after: float | None = None,
    ) -> None:
        self.retry_after = retry_after
        super().__init__(message=message, status_code=429)


class SimproNotFoundError(SimproAPIError):
    """Resource not found (HTTP 404)."""

    def __init__(self, message: str = "Resource not found") -> None:
        super().__init__(message=message, status_code=404)

What does this code do?

SimproRateLimitError fires when Simpro returns HTTP 429. It carries a retry_after value (in seconds) so calling code knows how long to wait before retrying.

SimproNotFoundError fires on HTTP 404. Having a dedicated class lets callers distinguish "resource doesn't exist" from other API errors with a simple except SimproNotFoundError clause.

  • Save src/simpro_client/exceptions.py.
  • Confirm the module imports cleanly by running:
python -c "from simpro_client.exceptions import SimproError, SimproAuthError, SimproAPIError, SimproRateLimitError, SimproNotFoundError; print('All exceptions imported')"

You should see All exceptions imported printed with no errors.

Seeing an ImportError?

  • Make sure you are in the phase3 directory with your virtual environment activated.
  • Check that the filename is exactly exceptions.py inside src/simpro_client/.
  • Verify your editable install is still intact by running pip install -e "." again.

help me debug an ImportError for simpro_client.exceptions

Build the authentication manager

The AuthManager class is the single place that knows how to get a valid token. Every future HTTP request passes through it, so the rest of the codebase never thinks about auth mechanics.

  • Create a new file src/simpro_client/auth.py and add the class skeleton with its constructor and public entry point:
"""OAuth2 Client Credentials authentication manager."""

import time
from typing import Any

import httpx

from simpro_client.config import SimproSettings
from simpro_client.exceptions import SimproAuthError


class AuthManager:
    """Manages OAuth2 token lifecycle.

    Supports two modes:
    - client_credentials: Obtains tokens from the OAuth2 token endpoint
    - api_key: Uses a static Bearer token from configuration
    """

    def __init__(self, settings: SimproSettings) -> None:
        self._settings = settings
        self._access_token: str | None = None
        self._token_expiry: float = 0.0
        self._http_client = httpx.Client(timeout=settings.timeout)

What does this code do?

The constructor stores the settings, initialises an empty token cache (_access_token and _token_expiry), and creates a dedicated httpx.Client for token requests. Having its own HTTP client means token fetches are independent of the main API client's lifecycle.

  • Below the __init__ method, add the routing logic that decides which auth strategy to use:
    def get_token(self) -> str:
        """Return a valid access token, refreshing if necessary."""
        if self._settings.auth_mode == "api_key":
            return self._get_api_key_token()
        return self._get_oauth_token()

    def _get_api_key_token(self) -> str:
        """Return the static API key token."""
        if not self._settings.api_key:
            raise SimproAuthError("API key mode selected but no api_key configured")
        return self._settings.api_key

    def _get_oauth_token(self) -> str:
        """Return a cached token or fetch a new one."""
        if self._access_token and time.time() < self._token_expiry:
            return self._access_token
        return self._refresh_token()

What does this code do?

get_token() is the single public method callers use. It checks auth_mode from settings and delegates to the correct strategy.

_get_oauth_token() implements token caching. It only calls _refresh_token() when the cache is empty or the token has expired. This avoids hitting the token endpoint on every single API call.

  • Save auth.py and test that the API key path works by running:
python -c "
from simpro_client.config import SimproSettings
from simpro_client.auth import AuthManager
s = SimproSettings(
    base_url='http://test:8000/api/v1.0',
    token_url='http://test:8000/oauth2/token',
    client_id='x', client_secret='x',
    auth_mode='api_key', api_key='my-test-key'
)
auth = AuthManager(s)
print(auth.get_token())
auth.close()
"

You should see my-test-key printed. The API key path works without needing a real token endpoint.

Seeing an AttributeError about close()?

The close() method does not exist yet. If you see this error, it confirms your code loaded correctly. You will add close() in the next chunk. Remove the auth.close() line from the test command and re-run to verify my-test-key prints.

help me debug an AttributeError in my AuthManager class

  • Now add the core OAuth2 refresh logic below _get_oauth_token:
    def _refresh_token(self) -> str:
        """Fetch a new token from the OAuth2 endpoint."""
        payload: dict[str, Any] = {
            "grant_type": "client_credentials",
            "client_id": self._settings.client_id,
            "client_secret": self._settings.client_secret,
        }
        try:
            response = self._http_client.post(
                self._settings.token_url,
                data=payload,
                headers={"Content-Type": "application/x-www-form-urlencoded"},
            )
        except httpx.HTTPError as e:
            raise SimproAuthError(f"Token request failed: {e}") from e

        if response.status_code != 200:
            raise SimproAuthError(
                f"Token endpoint returned {response.status_code}: {response.text}"
            )

        data = response.json()
        self._access_token = data["access_token"]
        expires_in = data.get("expires_in", 3600)
        self._token_expiry = time.time() + expires_in - 60
        return self._access_token

What does this code do?

This method POSTs the Client Credentials grant to the token endpoint. The payload includes grant_type, client_id, and client_secret as form-encoded data.

On success, it caches the access_token and calculates an expiry timestamp 60 seconds before the real expiry. That 60-second buffer avoids using a token that is about to expire mid-request.

  • Finally, add the lifecycle utility methods below _refresh_token:
    def invalidate(self) -> None:
        """Force token refresh on next call."""
        self._access_token = None
        self._token_expiry = 0.0

    def close(self) -> None:
        """Close the internal HTTP client."""
        self._http_client.close()
  • Save src/simpro_client/auth.py.
  • Confirm the full module passes linting by running:
ruff check src/simpro_client/auth.py

You should see no output, meaning zero errors and zero warnings.

Seeing linting errors?

  • An unused import warning means you have from typing import Any but _refresh_token is missing or uses a different type hint. Verify the payload: dict[str, Any] annotation is present.
  • An indentation error means a method is not indented inside the class body. Every def should be indented 4 spaces under class AuthManager.

help me fix ruff linting errors in my auth.py

✔️ Awesome, I've got everything!

Great. Double check you have saved both exceptions.py and auth.py.

ⓧ I'd like to double check the full code

For src/simpro_client/exceptions.py, confirm these sections appear in order:

  • Module docstring.
  • SimproError base class with message attribute.
  • SimproAuthError inheriting from SimproError.
  • SimproAPIError with status_code and response_body.
  • SimproRateLimitError with retry_after.
  • SimproNotFoundError defaulting to status 404.

For src/simpro_client/auth.py, confirm these sections appear in order:

  • Imports (time, Any, httpx, SimproSettings, SimproAuthError).
  • AuthManager class with __init__ storing settings, token cache, and HTTP client.
  • get_token() routing by auth_mode.
  • _get_api_key_token() and _get_oauth_token() helper methods.
  • _refresh_token() with the full POST + cache logic.
  • invalidate() and close() at the end.
Write the auth tests

The auth module talks to a remote token endpoint, but your tests should never depend on a live server. You will use respx to intercept all httpx calls and return controlled responses.

  • Create tests/test_auth.py and add the imports plus the first test (successful token retrieval):
"""Tests for authentication module."""

import time

import respx
from httpx import Response

from simpro_client.auth import AuthManager
from simpro_client.config import SimproSettings
from simpro_client.exceptions import SimproAuthError

import pytest


@respx.mock
def test_token_obtained_on_first_call(mock_settings: SimproSettings):
    respx.post(mock_settings.token_url).mock(
        return_value=Response(200, json={"access_token": "fresh-token", "expires_in": 3600})
    )
    auth = AuthManager(mock_settings)
    token = auth.get_token()
    assert token == "fresh-token"
    auth.close()

What does this code do?

The @respx.mock decorator intercepts all httpx traffic for the duration of the test. respx.post(mock_settings.token_url).mock(...) tells it to return a fake 200 response with a known token when the auth manager POSTs to the token URL.

The test uses the mock_settings fixture from conftest.py you created in the previous step. No real HTTP traffic ever leaves your machine.

  • Save and run this single test to confirm the test infrastructure works:
pytest tests/test_auth.py -v

You should see 1 test collected and PASSED in green.

Test failing with a connection error?

  • If you see a real HTTP connection attempt, the @respx.mock decorator is missing from the test function.
  • If the fixture is not found, confirm tests/conftest.py exists with the mock_settings fixture.

help me debug a failing respx test in test_auth.py

  • Below the first test, add the caching and refresh tests:
@respx.mock
def test_token_cached_on_second_call(mock_settings: SimproSettings):
    route = respx.post(mock_settings.token_url).mock(
        return_value=Response(200, json={"access_token": "cached-token", "expires_in": 3600})
    )
    auth = AuthManager(mock_settings)
    auth.get_token()
    auth.get_token()
    assert route.call_count == 1
    auth.close()


@respx.mock
def test_expired_token_triggers_refresh(mock_settings: SimproSettings):
    route = respx.post(mock_settings.token_url).mock(
        return_value=Response(200, json={"access_token": "refreshed-token", "expires_in": 3600})
    )
    auth = AuthManager(mock_settings)
    auth.get_token()
    auth._token_expiry = time.time() - 1
    token = auth.get_token()
    assert token == "refreshed-token"
    assert route.call_count == 2
    auth.close()

What does this code do?

The caching test calls get_token() twice and asserts the mock endpoint was only hit once. The second call returns the cached value.

The refresh test manually sets _token_expiry to a past timestamp, simulating an expired token. The next get_token() call must hit the endpoint again (call_count == 2).

  • Add the error handling test and the API key mode test at the end of the file:
@respx.mock
def test_invalid_credentials_raise_auth_error(mock_settings: SimproSettings):
    respx.post(mock_settings.token_url).mock(
        return_value=Response(401, text="Invalid client credentials")
    )
    auth = AuthManager(mock_settings)
    with pytest.raises(SimproAuthError, match="401"):
        auth.get_token()
    auth.close()


def test_api_key_mode_returns_static_token(api_key_settings: SimproSettings):
    auth = AuthManager(api_key_settings)
    token = auth.get_token()
    assert token == "test-static-token"
    auth.close()

What does this code do?

The invalid credentials test mocks a 401 response and confirms SimproAuthError is raised with the status code in the message.

The API key test uses the api_key_settings fixture (which sets auth_mode="api_key"). It confirms the static token is returned directly without any HTTP call. Notice there is no @respx.mock decorator since no network call should happen.

  • Save tests/test_auth.py.

Before you run the tests, what do you expect to see? Five tests passing, or a failure?

  • Run the full auth test suite:
pytest tests/test_auth.py -v

You should see 5 tests collected and all passing in green:

tests/test_auth.py::test_token_obtained_on_first_call PASSED
tests/test_auth.py::test_token_cached_on_second_call PASSED
tests/test_auth.py::test_expired_token_triggers_refresh PASSED
tests/test_auth.py::test_invalid_credentials_raise_auth_error PASSED
tests/test_auth.py::test_api_key_mode_returns_static_token PASSED

One or more tests failing?

  • If the caching test fails with call_count == 2, your _get_oauth_token method is not checking _token_expiry before refreshing.
  • If the refresh test fails, confirm you set _token_expiry using time.time() + expires_in - 60 in _refresh_token.
  • If the API key test fails with an HTTP error, check that get_token() checks auth_mode before trying the OAuth flow.

help me debug failing auth tests in my simpro_client project

✔️ Awesome, I've got everything!

Great. Confirm you have saved tests/test_auth.py and all 5 tests pass.

ⓧ I'd like to double check the full code

For tests/test_auth.py, confirm these test functions appear in order:

  • test_token_obtained_on_first_call (mocks 200, asserts "fresh-token").
  • test_token_cached_on_second_call (calls twice, asserts call_count == 1).
  • test_expired_token_triggers_refresh (forces expiry, asserts call_count == 2).
  • test_invalid_credentials_raise_auth_error (mocks 401, asserts SimproAuthError).
  • test_api_key_mode_returns_static_token (no mock decorator, asserts "test-static-token").

Your authentication module is complete. Every future endpoint module will call get_token() and receive a valid Bearer token without worrying about expiry, refresh, or auth mode. Next up, you will build the base HTTP client that uses this auth layer to send real API requests with proper error handling.

Build the Base HTTP Client

Your auth module can obtain and cache tokens. Your exception hierarchy gives downstream code specific errors to catch.

Now every future endpoint module needs the same boilerplate: inject the Bearer token, set timeouts, parse errors, retry on 401. Without a centralized client, each endpoint duplicates all of that logic. The base HTTP client solves this once so that every endpoint you build in Sprint 3 inherits correct auth and error handling automatically.

In this step, get ready to:
  • Create a logging utility module with correlation IDs and request timing.
  • Build a client class that wraps httpx with auth injection and typed error responses.
  • Write integration-style tests proving the full request cycle with mocked HTTP responses.
Create the logging utility module

The client needs structured logging to trace requests. You will create a small utility module that the client imports for correlation IDs and request timing.

  • Create a new file called src/simpro_client/logging.py and add the imports and correlation ID helpers:
"""Structured logging configuration with correlation ID support."""

import json
import logging
import time
import uuid
from contextvars import ContextVar
from typing import Any

# Thread-safe variable that holds the current correlation ID
correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")


def get_correlation_id() -> str:
    """Get the current correlation ID or generate a new one."""
    cid = correlation_id_var.get()
    if not cid:
        cid = uuid.uuid4().hex[:12]
        correlation_id_var.set(cid)
    return cid


def set_correlation_id(cid: str | None = None) -> str:
    """Set a correlation ID for the current context."""
    if cid is None:
        cid = uuid.uuid4().hex[:12]
    correlation_id_var.set(cid)
    return cid

What does this code do?

correlation_id_var is a ContextVar that stores a unique ID for the current logical operation. It is thread-safe and async-safe.

get_correlation_id() returns the current ID or auto-generates a 12-character hex string if none exists. set_correlation_id() lets you explicitly set one for a request chain.

  • Below the set_correlation_id function, add the JSON log formatter class:

class JSONFormatter(logging.Formatter):
    """Format log records as JSON with correlation IDs."""

    def format(self, record: logging.LogRecord) -> str:
        log_data: dict[str, Any] = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "correlation_id": correlation_id_var.get(""),
        }
        # Include HTTP fields when present on the log record
        if hasattr(record, "method"):
            log_data["method"] = record.method
        if hasattr(record, "url"):
            log_data["url"] = record.url
        if hasattr(record, "status_code"):
            log_data["status_code"] = record.status_code
        if hasattr(record, "duration_ms"):
            log_data["duration_ms"] = record.duration_ms
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data)

What does this code do?

JSONFormatter outputs each log record as a single JSON object. It always includes the correlation ID and timestamp.

When HTTP-specific attributes (method, url, status_code, duration_ms) are attached to the record, they appear in the JSON output. This makes logs machine-parseable for debugging.

  • Verify the module is importable so far by running:
python -c "from simpro_client.logging import get_correlation_id; print(get_correlation_id())"

You should see a 12-character hex string printed. This confirms the correlation ID auto-generation works.

  • Back in src/simpro_client/logging.py, add the logger setup function and request timer below the JSONFormatter class:


def configure_logging(level: str = "INFO") -> logging.Logger:
    """Configure structured JSON logging for the simpro_client package."""
    logger = logging.getLogger("simpro_client")
    logger.setLevel(getattr(logging, level.upper(), logging.INFO))
    if not logger.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(JSONFormatter())
        logger.addHandler(handler)
    return logger


class RequestTimer:
    """Context manager for timing HTTP requests."""

    def __init__(self) -> None:
        self.start_time: float = 0.0
        self.duration_ms: float = 0.0

    def __enter__(self) -> "RequestTimer":
        self.start_time = time.perf_counter()
        return self

    def __exit__(self, *args: Any) -> None:
        self.duration_ms = (time.perf_counter() - self.start_time) * 1000

What does this code do?

configure_logging() creates a logger named simpro_client with JSON output. The if not logger.handlers guard prevents duplicate handlers if called multiple times.

RequestTimer is a context manager that measures elapsed time in milliseconds. The client wraps each HTTP call in it to log duration.

  • Save src/simpro_client/logging.py.
  • Verify the full module loads by running:
python -c "from simpro_client.logging import configure_logging, RequestTimer; print('logging module OK')"

You should see logging module OK printed.

Seeing an ImportError?

Check that the file is saved at the exact path src/simpro_client/logging.py (not src/logging.py). Also confirm your virtual environment is active. help me fix this ImportError in my logging module

Build the client class with response handling

The client wraps httpx.Client with auth injection, timeout configuration, and typed error mapping. Every endpoint module in Sprint 3 will call through this class.

  • Create a new file called src/simpro_client/client.py and add the imports and constructor:
"""Base HTTP client for the Simpro REST API."""

import logging
from typing import Any

import httpx

from simpro_client.auth import AuthManager
from simpro_client.config import SimproSettings, get_settings
from simpro_client.exceptions import (
    SimproAPIError,
    SimproAuthError,
    SimproNotFoundError,
    SimproRateLimitError,
)
from simpro_client.logging import (
    RequestTimer,
    configure_logging,
    get_correlation_id,
    set_correlation_id,
)


class SimproClient:
    """Main entry point for the Simpro API integration."""

    def __init__(self, settings: SimproSettings | None = None) -> None:
        self._settings = settings or get_settings()
        self._auth = AuthManager(self._settings)
        self._logger = configure_logging()
        # Configure httpx with base URL, timeout, and default headers
        self._http = httpx.Client(
            base_url=self._settings.base_url,
            timeout=self._settings.timeout,
            headers={"Content-Type": "application/json", "Accept": "application/json"},
        )

What does this code do?

The constructor wires together the pieces you built in previous steps: AuthManager for token management, configure_logging() for structured output, and httpx.Client for connection pooling.

The base_url setting means all requests use relative paths (like /companies/1/jobs/). The timeout comes from your config module.

  • Below the constructor, add the public methods and the core request execution logic:

    def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
        """Send a GET request."""
        return self._request("GET", path, params=params)

    def post(self, path: str, json: dict[str, Any] | None = None) -> Any:
        """Send a POST request."""
        return self._request("POST", path, json=json)

    def patch(self, path: str, json: dict[str, Any] | None = None) -> Any:
        """Send a PATCH request."""
        return self._request("PATCH", path, json=json)

    def delete(self, path: str) -> Any:
        """Send a DELETE request."""
        return self._request("DELETE", path)

    def _request(
        self,
        method: str,
        path: str,
        params: dict[str, Any] | None = None,
        json: dict[str, Any] | None = None,
        _retry_on_401: bool = True,
    ) -> Any:
        """Execute an HTTP request with auth, logging, and error handling."""
        cid = get_correlation_id() or set_correlation_id()
        token = self._auth.get_token()
        headers = {"Authorization": f"Bearer {token}"}

What does this code do?

The four public methods (get, post, patch, delete) delegate to _request() which handles auth injection.

_request() starts by ensuring a correlation ID exists for tracing, then fetches a valid token from AuthManager and builds the Authorization header.

  • Verify the file is syntactically valid so far by running:
python -c "import ast; ast.parse(open('src/simpro_client/client.py').read()); print('syntax OK')"

You should see syntax OK. The module is not fully importable yet because _request() is incomplete.

  • Continue _request() by adding the HTTP execution and then the response handler directly below the headers = {"Authorization": ...} line:

        with RequestTimer() as timer:
            try:
                response = self._http.request(
                    method=method, url=path, params=params, json=json, headers=headers
                )
            except httpx.HTTPError as e:
                self._logger.error(f"Request failed: {method} {path}: {e}", extra={"method": method, "url": path})
                raise SimproAPIError(message=f"Request failed: {e}", status_code=0) from e

        self._log_request(method, path, response.status_code, timer.duration_ms)
        return self._handle_response(response, method, path, params, json, _retry_on_401)

    def _handle_response(
        self,
        response: httpx.Response,
        method: str,
        path: str,
        params: dict[str, Any] | None,
        json: dict[str, Any] | None,
        retry_on_401: bool,
    ) -> Any:
        """Process the response, handling errors and retries."""
        if response.status_code == 401 and retry_on_401:
            self._logger.info("Received 401, refreshing token and retrying")
            self._auth.invalidate()
            return self._request(method, path, params=params, json=json, _retry_on_401=False)
        if response.status_code == 404:
            raise SimproNotFoundError(f"Not found: {method} {path}")
        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After")
            raise SimproRateLimitError(retry_after=float(retry_after) if retry_after else None)
        if response.status_code >= 400:
            raise SimproAPIError(
                message=f"API error: {response.status_code} on {method} {path}",
                status_code=response.status_code,
                response_body=response.text,
            )

What does this code do?

RequestTimer wraps the HTTP call to measure duration in milliseconds. If the network call itself fails, it raises SimproAPIError with status_code=0.

_handle_response() maps HTTP status codes to your typed exceptions: 401 triggers one token refresh and retry, 404 raises SimproNotFoundError, 429 raises SimproRateLimitError (preserving the Retry-After header), and other 4xx/5xx raise SimproAPIError.

  • Finish _handle_response() and add the remaining methods. Paste this directly below the response_body=response.text, block:
        if response.status_code == 204:
            return None
        return response.json()

    def _log_request(self, method: str, url: str, status_code: int, duration_ms: float) -> None:
        """Log request details with structured fields."""
        log_record = logging.LogRecord(
            name="simpro_client", level=logging.INFO, pathname="", lineno=0,
            msg=f"{method} {url} -> {status_code} ({duration_ms:.1f}ms)", args=None, exc_info=None,
        )
        log_record.method = method
        log_record.url = url
        log_record.status_code = status_code
        log_record.duration_ms = round(duration_ms, 1)
        self._logger.handle(log_record)

    def close(self) -> None:
        """Close all underlying connections."""
        self._http.close()
        self._auth.close()

    def __enter__(self) -> "SimproClient":
        return self

    def __exit__(self, *args: Any) -> None:
        self.close()

What does this code do?

204 (No Content) returns None instead of trying to parse an empty body. Any 2xx with a body returns parsed JSON.

_log_request() builds a structured LogRecord with HTTP fields attached. The context manager methods (__enter__/__exit__) let you use with SimproClient() as client: for automatic cleanup.

  • Save src/simpro_client/client.py.
  • Verify the full module imports correctly by running:
python -c "from simpro_client.client import SimproClient; print('client module OK')"

You should see client module OK printed. The package now has a working client class.

Seeing a NameError or IndentationError?

Check that all methods are indented one level inside the class SimproClient: block (4 spaces). Also confirm the closing parenthesis of the SimproAPIError raise in _handle_response ends with a comma after response_body=response.text. help me fix this error in my client module

Write integration tests for the client

You will use respx to mock HTTP responses and prove the full request cycle: auth, request execution, error handling, and cleanup.

  • Create tests/test_client.py and add the imports and first two test cases:
"""Tests for the base HTTP client."""

import respx
from httpx import Response

from simpro_client.client import SimproClient
from simpro_client.config import SimproSettings
from simpro_client.exceptions import SimproAPIError, SimproNotFoundError, SimproRateLimitError

import pytest


@respx.mock
def test_successful_get(mock_settings):
    """Successful GET returns parsed JSON."""
    respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/").mock(return_value=Response(200, json=[{"ID": 1, "Name": "Test Job"}]))
    with SimproClient(settings=mock_settings) as client:
        data = client.get("/companies/1/jobs/")
    assert data == [{"ID": 1, "Name": "Test Job"}]


@respx.mock
def test_401_triggers_refresh_and_retry(mock_settings):
    """401 response triggers token refresh and retries the request."""
    token_route = respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "new-tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/").mock(side_effect=[Response(401, text="Unauthorized"), Response(200, json=[{"ID": 1}])])
    with SimproClient(settings=mock_settings) as client:
        data = client.get("/companies/1/jobs/")
    assert data == [{"ID": 1}]
    assert token_route.call_count == 2

What do these tests prove?

test_successful_get mocks both the token endpoint and a jobs endpoint. It proves the client fetches a token, injects it as a Bearer header, and returns parsed JSON.

test_401_triggers_refresh_and_retry returns a 401 on the first attempt. It proves the client invalidates the token, fetches a new one (hence call_count == 2), and retries successfully.

  • Below those tests, add the remaining error-case tests and the context manager test:


@respx.mock
def test_404_raises_not_found(mock_settings):
    """404 response raises SimproNotFoundError."""
    respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/999").mock(return_value=Response(404, text="Not Found"))
    with SimproClient(settings=mock_settings) as client:
        with pytest.raises(SimproNotFoundError):
            client.get("/companies/1/jobs/999")


@respx.mock
def test_429_raises_rate_limit_error(mock_settings):
    """429 response raises SimproRateLimitError with retry_after."""
    respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/").mock(return_value=Response(429, headers={"Retry-After": "5"}))
    with SimproClient(settings=mock_settings) as client:
        with pytest.raises(SimproRateLimitError) as exc_info:
            client.get("/companies/1/jobs/")
    assert exc_info.value.retry_after == 5.0


@respx.mock
def test_client_context_manager(mock_settings):
    """Client works as a context manager and closes cleanly."""
    respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/").mock(return_value=Response(200, json={"ID": 1, "Name": "CVC Service"}))
    with SimproClient(settings=mock_settings) as client:
        result = client.get("/companies/1/")
    assert result["Name"] == "CVC Service"

What do these tests prove?

test_404_raises_not_found confirms the client maps a 404 to SimproNotFoundError (not a generic exception).

test_429_raises_rate_limit_error verifies the Retry-After header is parsed into the retry_after attribute. Simpro enforces a rate limit of 10 requests per second per build.

test_client_context_manager proves the with statement closes all connections cleanly.

  • Save tests/test_client.py.

Before you run the full test suite, what do you expect to see? All auth tests from earlier should still pass, plus 5 new client tests.

  • Run the full test suite by running:
pytest tests/ -v

You should see all tests passing, including the 5 new client tests and all previous auth and config tests.

  • Confirm linting and formatting pass by running:
ruff check src/ && ruff format --check src/

You should see zero errors and zero warnings from both commands.

Seeing test failures or ruff errors?

If a test fails with SimproAuthError, check that the respx.post mock uses the exact mock_settings.token_url value. The URL must match character-for-character.

If ruff reports an unused import, you can add a # noqa: F401 comment on the relevant import line. help me fix my test or linting failures

✔️ Awesome, I've got everything!

Great. Double check you have saved both src/simpro_client/logging.py and src/simpro_client/client.py.

ⓧ I'd like to double check the full code

Here is the full src/simpro_client/logging.py:

"""Structured logging configuration with correlation ID support."""

import json
import logging
import time
import uuid
from contextvars import ContextVar
from typing import Any

correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")


def get_correlation_id() -> str:
    cid = correlation_id_var.get()
    if not cid:
        cid = uuid.uuid4().hex[:12]
        correlation_id_var.set(cid)
    return cid


def set_correlation_id(cid: str | None = None) -> str:
    if cid is None:
        cid = uuid.uuid4().hex[:12]
    correlation_id_var.set(cid)
    return cid
class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_data: dict[str, Any] = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "correlation_id": correlation_id_var.get(""),
        }
        if hasattr(record, "method"):
            log_data["method"] = record.method
        if hasattr(record, "url"):
            log_data["url"] = record.url
        if hasattr(record, "status_code"):
            log_data["status_code"] = record.status_code
        if hasattr(record, "duration_ms"):
            log_data["duration_ms"] = record.duration_ms
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data)


def configure_logging(level: str = "INFO") -> logging.Logger:
    logger = logging.getLogger("simpro_client")
    logger.setLevel(getattr(logging, level.upper(), logging.INFO))
    if not logger.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(JSONFormatter())
        logger.addHandler(handler)
    return logger
class RequestTimer:
    def __init__(self) -> None:
        self.start_time: float = 0.0
        self.duration_ms: float = 0.0

    def __enter__(self) -> "RequestTimer":
        self.start_time = time.perf_counter()
        return self

    def __exit__(self, *args: Any) -> None:
        self.duration_ms = (time.perf_counter() - self.start_time) * 1000

Here is the full src/simpro_client/client.py:

"""Base HTTP client for the Simpro REST API."""

import logging
from typing import Any

import httpx

from simpro_client.auth import AuthManager
from simpro_client.config import SimproSettings, get_settings
from simpro_client.exceptions import (
    SimproAPIError,
    SimproAuthError,
    SimproNotFoundError,
    SimproRateLimitError,
)
from simpro_client.logging import (
    RequestTimer,
    configure_logging,
    get_correlation_id,
    set_correlation_id,
)
class SimproClient:
    def __init__(self, settings: SimproSettings | None = None) -> None:
        self._settings = settings or get_settings()
        self._auth = AuthManager(self._settings)
        self._logger = configure_logging()
        self._http = httpx.Client(
            base_url=self._settings.base_url,
            timeout=self._settings.timeout,
            headers={"Content-Type": "application/json", "Accept": "application/json"},
        )

    def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
        return self._request("GET", path, params=params)

    def post(self, path: str, json: dict[str, Any] | None = None) -> Any:
        return self._request("POST", path, json=json)

    def patch(self, path: str, json: dict[str, Any] | None = None) -> Any:
        return self._request("PATCH", path, json=json)

    def delete(self, path: str) -> Any:
        return self._request("DELETE", path)
    def _request(
        self,
        method: str,
        path: str,
        params: dict[str, Any] | None = None,
        json: dict[str, Any] | None = None,
        _retry_on_401: bool = True,
    ) -> Any:
        cid = get_correlation_id() or set_correlation_id()
        token = self._auth.get_token()
        headers = {"Authorization": f"Bearer {token}"}

        with RequestTimer() as timer:
            try:
                response = self._http.request(
                    method=method, url=path, params=params, json=json, headers=headers
                )
            except httpx.HTTPError as e:
                self._logger.error(f"Request failed: {method} {path}: {e}", extra={"method": method, "url": path})
                raise SimproAPIError(message=f"Request failed: {e}", status_code=0) from e

        self._log_request(method, path, response.status_code, timer.duration_ms)
        return self._handle_response(response, method, path, params, json, _retry_on_401)
    def _handle_response(
        self,
        response: httpx.Response,
        method: str,
        path: str,
        params: dict[str, Any] | None,
        json: dict[str, Any] | None,
        retry_on_401: bool,
    ) -> Any:
        if response.status_code == 401 and retry_on_401:
            self._logger.info("Received 401, refreshing token and retrying")
            self._auth.invalidate()
            return self._request(method, path, params=params, json=json, _retry_on_401=False)
        if response.status_code == 404:
            raise SimproNotFoundError(f"Not found: {method} {path}")
        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After")
            raise SimproRateLimitError(retry_after=float(retry_after) if retry_after else None)
        if response.status_code >= 400:
            raise SimproAPIError(
                message=f"API error: {response.status_code} on {method} {path}",
                status_code=response.status_code,
                response_body=response.text,
            )
        if response.status_code == 204:
            return None
        return response.json()

    def _log_request(self, method: str, url: str, status_code: int, duration_ms: float) -> None:
        log_record = logging.LogRecord(
            name="simpro_client", level=logging.INFO, pathname="", lineno=0,
            msg=f"{method} {url} -> {status_code} ({duration_ms:.1f}ms)", args=None, exc_info=None,
        )
        log_record.method = method
        log_record.url = url
        log_record.status_code = status_code
        log_record.duration_ms = round(duration_ms, 1)
        self._logger.handle(log_record)

    def close(self) -> None:
        self._http.close()
        self._auth.close()

    def __enter__(self) -> "SimproClient":
        return self

    def __exit__(self, *args: Any) -> None:
        self.close()

Here is the full tests/test_client.py:

"""Tests for the base HTTP client."""

import respx
from httpx import Response

from simpro_client.client import SimproClient
from simpro_client.config import SimproSettings
from simpro_client.exceptions import SimproAPIError, SimproNotFoundError, SimproRateLimitError

import pytest


@respx.mock
def test_successful_get(mock_settings):
    respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/").mock(return_value=Response(200, json=[{"ID": 1, "Name": "Test Job"}]))
    with SimproClient(settings=mock_settings) as client:
        data = client.get("/companies/1/jobs/")
    assert data == [{"ID": 1, "Name": "Test Job"}]


@respx.mock
def test_401_triggers_refresh_and_retry(mock_settings):
    token_route = respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "new-tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/").mock(side_effect=[Response(401, text="Unauthorized"), Response(200, json=[{"ID": 1}])])
    with SimproClient(settings=mock_settings) as client:
        data = client.get("/companies/1/jobs/")
    assert data == [{"ID": 1}]
    assert token_route.call_count == 2
@respx.mock
def test_404_raises_not_found(mock_settings):
    respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/999").mock(return_value=Response(404, text="Not Found"))
    with SimproClient(settings=mock_settings) as client:
        with pytest.raises(SimproNotFoundError):
            client.get("/companies/1/jobs/999")


@respx.mock
def test_429_raises_rate_limit_error(mock_settings):
    respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/jobs/").mock(return_value=Response(429, headers={"Retry-After": "5"}))
    with SimproClient(settings=mock_settings) as client:
        with pytest.raises(SimproRateLimitError) as exc_info:
            client.get("/companies/1/jobs/")
    assert exc_info.value.retry_after == 5.0


@respx.mock
def test_client_context_manager(mock_settings):
    respx.post(mock_settings.token_url).mock(return_value=Response(200, json={"access_token": "tok", "expires_in": 3600}))
    respx.get(f"{mock_settings.base_url}/companies/1/").mock(return_value=Response(200, json={"ID": 1, "Name": "CVC Service"}))
    with SimproClient(settings=mock_settings) as client:
        result = client.get("/companies/1/")
    assert result["Name"] == "CVC Service"

Your base HTTP client is fully operational. Every future endpoint module inherits auth injection, timeout handling, and typed exceptions for free. Next up, you will tackle the Secret Mission: adding structured JSON logging with correlation IDs that trace every request through the system.

Secret mission

Add Structured Logging with Correlation IDs

Your client already has logging infrastructure wired in. But can you prove it works? Write a full test suite that verifies every log line includes structured JSON fields and correlation IDs, then document the feature so future developers know how to use it.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources in place for Sprint 2, pause them to free up disk space, or delete them entirely.

Resources you used:

  • Python virtual environment (.venv inside phase3/).
  • Project source files (phase3/src/, phase3/tests/, phase3/pyproject.toml).
  • Environment file (.env with your Simpro connection settings).

Keep everything running

No action needed. Choose this if you are continuing to Sprint 2, where you will build the mock API that this client connects to.

  • Leave the phase3/ directory and virtual environment in place.
  • Your .env file stays where it is. It is already in .gitignore so credentials will not be committed.

Pause - I'll come back to this later

Remove the virtual environment to free up disk space while keeping all your source files intact.

  • Delete the virtual environment by running:
rm -rf phase3/.venv
  • When you return, recreate it by running:
cd phase3
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
  • Your source files, tests, and .env remain untouched.

Delete - I don't want to use this again

Remove the entire project directory and all its contents.

  • Delete the phase3/ directory by running:
rm -rf phase3/
  • Confirm deletion by running:
ls phase3/
  • You should see a "No such file or directory" message, confirming the folder is gone.
  • This removes all source code, tests, the virtual environment, and your .env file. You would need to redo the entire sprint to get it back.

Nice Work!

Nice Work!

You just built a production-grade Python integration layer from scratch. The simpro_client package handles the two hardest parts of any API integration: authentication and HTTP transport.

You've learned how to:

  • Build a validated configuration module using Pydantic that catches misconfigurations at startup and loads settings from environment variables.
  • Build an OAuth2 Client Credentials auth manager that obtains, caches, and automatically refreshes tokens without manual intervention.
  • Build a base HTTP client with Bearer token injection, connection pooling, typed exception handling, and automatic 401 retry using httpx.
  • Secret Mission: Added structured JSON logging with correlation IDs that trace every request through the system for production debugging.

Ready to quiz yourself?