Skip to content

API Reference

Stages

Base Classes

base

Base classes for stages.

Classes

Status

Bases: Enum

Check result status.

Source code in src/uptimer/stages/base.py 
 9
10
11
12
13
14
class Status(Enum):
    """Check result status."""

    UP = "up"
    DEGRADED = "degraded"
    DOWN = "down"

CheckResult dataclass

Result of a check.

Source code in src/uptimer/stages/base.py 
17
18
19
20
21
22
23
24
25
@dataclass
class CheckResult:
    """Result of a check."""

    status: Status
    url: str
    message: str
    elapsed_ms: float = 0.0
    details: dict[str, Any] = field(default_factory=lambda: {})

CheckContext dataclass

Context passed between stages in a pipeline.

Source code in src/uptimer/stages/base.py 
28
29
30
31
32
33
34
35
36
37
@dataclass
class CheckContext:
    """Context passed between stages in a pipeline."""

    url: str
    response_body: str | None = None
    response_headers: dict[str, str] = field(default_factory=lambda: {})
    status_code: int | None = None
    values: dict[str, Any] = field(default_factory=lambda: {})
    elapsed_ms: float = 0.0

Stage

Bases: ABC

Base class for all stages.

Source code in src/uptimer/stages/base.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class Stage(ABC):
    """Base class for all stages."""

    name: str = "base"
    description: str = "Base stage"

    # Whether this stage makes HTTP requests (vs transforms data)
    is_network_stage: bool = True

    @abstractmethod
    def check(self, url: str, verbose: bool = False, context: CheckContext | None = None) -> CheckResult:
        """Perform the check and return result.

        Args:
            url: URL to check
            verbose: Whether to include verbose output
            context: Optional context from previous stages

        Returns:
            CheckResult with status, message, and details
        """
        pass
Functions
check(url, verbose=False, context=None) abstractmethod

Perform the check and return result.

Parameters:

Name Type Description Default
url str

URL to check

 required
verbose bool

Whether to include verbose output

 False
context CheckContext | None

Optional context from previous stages

 None

Returns:

Type Description
CheckResult

CheckResult with status, message, and details
Source code in src/uptimer/stages/base.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
@abstractmethod
def check(self, url: str, verbose: bool = False, context: CheckContext | None = None) -> CheckResult:
    """Perform the check and return result.

    Args:
        url: URL to check
        verbose: Whether to include verbose output
        context: Optional context from previous stages

    Returns:
        CheckResult with status, message, and details
    """
    pass

HTTP Stage

HttpStage

Bases: Stage

HTTP stage that follows redirects.

Source code in src/uptimer/stages/http.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class HttpStage(Stage):
    """HTTP stage that follows redirects."""

    name = "http"
    description = "HTTP check with redirect following"
    is_network_stage = True

    # User-Agent to avoid being blocked by sites that reject bot traffic
    USER_AGENT = "Mozilla/5.0 (compatible; Uptimer/1.0; +https://github.com/mortenoh/uptimer)"

    def __init__(self, timeout: float = 10.0, headers: dict[str, str] | None = None) -> None:
        """Initialize with timeout and optional custom headers.

        Args:
            timeout: Request timeout in seconds
            headers: Custom HTTP headers to send with the request
        """
        self.timeout = timeout
        self.custom_headers = headers or {}

    def check(self, url: str, verbose: bool = False, context: CheckContext | None = None) -> CheckResult:
        """Check URL via HTTP GET, following redirects."""
        # Add https:// if no protocol specified
        if not url.startswith(("http://", "https://")):
            url = f"https://{url}"

        details: dict[str, object] = {}

        try:
            start = time.perf_counter()
            headers = {"User-Agent": self.USER_AGENT, **self.custom_headers}
            with httpx.Client(timeout=self.timeout, follow_redirects=True, headers=headers) as client:
                response = client.get(url)
                elapsed_ms = (time.perf_counter() - start) * 1000

                # Determine status
                if response.status_code < 400:
                    status = Status.UP
                else:
                    status = Status.DEGRADED

                # Build details
                details["status_code"] = response.status_code
                details["http_version"] = response.http_version
                details["final_url"] = str(response.url)

                if response.headers.get("server"):
                    details["server"] = response.headers["server"]
                if response.headers.get("content-type"):
                    details["content_type"] = response.headers["content-type"]

                # Redirect chain
                if response.history:
                    details["redirects"] = [
                        {"status": r.status_code, "location": r.headers.get("location", "")} for r in response.history
                    ]

                # Store response data in context for subsequent stages
                if context is not None:
                    context.response_body = response.text
                    context.response_headers = dict(response.headers)
                    context.status_code = response.status_code
                    context.elapsed_ms = elapsed_ms

                return CheckResult(
                    status=status,
                    url=url,
                    message=str(response.status_code),
                    elapsed_ms=elapsed_ms,
                    details=details,
                )

        except httpx.RequestError as e:
            return CheckResult(
                status=Status.DOWN,
                url=url,
                message=e.__class__.__name__,
                details={"error": str(e)},
            )

Functions

__init__(timeout=10.0, headers=None)

Initialize with timeout and optional custom headers.

Parameters:

Name Type Description Default
timeout float

Request timeout in seconds

 10.0
headers dict[str, str] | None

Custom HTTP headers to send with the request

 None
Source code in src/uptimer/stages/http.py 
20
21
22
23
24
25
26
27
28
def __init__(self, timeout: float = 10.0, headers: dict[str, str] | None = None) -> None:
    """Initialize with timeout and optional custom headers.

    Args:
        timeout: Request timeout in seconds
        headers: Custom HTTP headers to send with the request
    """
    self.timeout = timeout
    self.custom_headers = headers or {}

check(url, verbose=False, context=None)

Check URL via HTTP GET, following redirects.

Source code in src/uptimer/stages/http.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def check(self, url: str, verbose: bool = False, context: CheckContext | None = None) -> CheckResult:
    """Check URL via HTTP GET, following redirects."""
    # Add https:// if no protocol specified
    if not url.startswith(("http://", "https://")):
        url = f"https://{url}"

    details: dict[str, object] = {}

    try:
        start = time.perf_counter()
        headers = {"User-Agent": self.USER_AGENT, **self.custom_headers}
        with httpx.Client(timeout=self.timeout, follow_redirects=True, headers=headers) as client:
            response = client.get(url)
            elapsed_ms = (time.perf_counter() - start) * 1000

            # Determine status
            if response.status_code < 400:
                status = Status.UP
            else:
                status = Status.DEGRADED

            # Build details
            details["status_code"] = response.status_code
            details["http_version"] = response.http_version
            details["final_url"] = str(response.url)

            if response.headers.get("server"):
                details["server"] = response.headers["server"]
            if response.headers.get("content-type"):
                details["content_type"] = response.headers["content-type"]

            # Redirect chain
            if response.history:
                details["redirects"] = [
                    {"status": r.status_code, "location": r.headers.get("location", "")} for r in response.history
                ]

            # Store response data in context for subsequent stages
            if context is not None:
                context.response_body = response.text
                context.response_headers = dict(response.headers)
                context.status_code = response.status_code
                context.elapsed_ms = elapsed_ms

            return CheckResult(
                status=status,
                url=url,
                message=str(response.status_code),
                elapsed_ms=elapsed_ms,
                details=details,
            )

    except httpx.RequestError as e:
        return CheckResult(
            status=Status.DOWN,
            url=url,
            message=e.__class__.__name__,
            details={"error": str(e)},
        )

Registry

registry

Stage registry for pluggable stage system.

Functions

register_stage(stage_class)

Register a stage class. Can be used as decorator.

Source code in src/uptimer/stages/registry.py 
11
12
13
14
def register_stage(stage_class: type["Stage"]) -> type["Stage"]:
    """Register a stage class. Can be used as decorator."""
    _registry[stage_class.name] = stage_class
    return stage_class

get_stage(name)

Get a stage class by name.

Source code in src/uptimer/stages/registry.py 
17
18
19
20
21
22
def get_stage(name: str) -> type["Stage"]:
    """Get a stage class by name."""
    if name not in _registry:
        available = ", ".join(_registry.keys())
        raise ValueError(f"Unknown stage: {name}. Available: {available}")
    return _registry[name]

list_stages()

List all registered stage names.

Source code in src/uptimer/stages/registry.py 
25
26
27
def list_stages() -> list[str]:
    """List all registered stage names."""
    return list(_registry.keys())

Logging

logging

Logging configuration using structlog.

Functions

configure_logging(json_output=False)

Configure structlog for console or JSON output.

Source code in src/uptimer/logging.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def configure_logging(json_output: bool = False) -> None:
    """Configure structlog for console or JSON output."""
    if json_output:
        # JSON output for metrics/machine consumption
        processors: list[structlog.typing.Processor] = [
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer(),
        ]
    else:
        # Human-readable console output
        processors = [
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="%H:%M:%S"),
            structlog.dev.ConsoleRenderer(colors=sys.stdout.isatty()),
        ]

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger(0),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

get_logger(name=None)

Get a logger instance.

Source code in src/uptimer/logging.py 
36
37
38
def get_logger(name: str | None = None) -> structlog.stdlib.BoundLogger:
    """Get a logger instance."""
    return structlog.get_logger(name)  # type: ignore[no-any-return]