Handling File and Network Failures with Targeted Exceptions and Cleanup

Good to see you again. Last lesson built a clean file-I/O boundary: Path objects identify artifacts, with scopes open resources, and JSON/JSONL/CSV parsing is followed by schema validation. That handles the happy path. Production AI systems also encounter deleted manifests, unreadable cache directories, malformed artifacts, unavailable model endpoints, slow upstream services, and HTTP responses that are technically valid but operationally unusable.

This lesson makes those boundaries reliable. You will distinguish operational failures from malformed data and programming errors; catch only the exceptions for which your code has a meaningful response; preserve the original cause when translating an error; and ensure files, responses, and temporary artifacts are cleaned up.

---

Exceptions are part of an interface contract

An exception is not merely a crash to suppress. It is Python’s structured signal that an operation could not fulfill its contract.

For an AI application, consider a function that loads an index manifest:

• A missing manifest might mean first-run setup is needed.
• A permission failure might mean a deployment configuration problem.
• Invalid JSON means the artifact was corrupted or manually edited incorrectly.
• A Pydantic validation failure means the JSON is valid but does not satisfy your application schema.
• An AttributeError inside your own implementation likely indicates a programming defect and should surface during development and testing.

Those cases require different actions. Treating them all as “something went wrong” loses the information needed to recover safely.

The core mechanism is try/except:

try:
    value = risky_operation()
except SpecificError as error:
    recover_or_translate(error)

The key word is specific. An except clause matches a class and its subclasses, and Python executes only the first matching handler. Therefore, put narrow subclasses before broad parent classes.

try:
    path.read_text(encoding="utf-8")
except FileNotFoundError:
    print("The artifact does not exist.")
except OSError:
    print("A different operating-system I/O error occurred.")

FileNotFoundError is an OSError subclass. Reversing these handlers would make the FileNotFoundError branch unreachable in practice because OSError would match first.

Do not use this as a default:

try:
    load_and_process_everything()
except Exception:
    return None

It discards the reason for failure and may convert a genuine coding bug into an apparently normal result. except Exception also makes debugging and observability much harder. It does not catch everything—KeyboardInterrupt and SystemExit derive from BaseException, not Exception—and that is intentional: users should generally be able to stop a process.

A practical rule is:

Catch an exception only when you can recover, add useful context, select an intentional fallback, or translate a low-level failure into a stable application-level contract.

---

Keep the protected region small

The size of a try block matters. It should contain only the operation whose failures you intend to handle.

Suppose a function loads a JSON manifest. There are three distinct failure categories:

1. Access failures: missing file, permission denied, disk or filesystem problems.
2. Representation failures: invalid UTF-8 or invalid JSON.
3. Schema failures: valid JSON that does not meet your application’s requirements.

The first two occur while reading and parsing. The third occurs when Pydantic validates the resulting object. Keep those boundaries visible:

import json
from pathlib import Path
from typing import Any


class ArtifactUnavailableError(Exception):
    """An artifact could not be accessed."""


class ArtifactContentError(Exception):
    """An artifact was accessible but malformed."""


def load_manifest_json(path: Path) -> dict[str, Any]:
    try:
        with path.open("r", encoding="utf-8") as file:
            raw: object = json.load(file)

    except FileNotFoundError as error:
        raise ArtifactUnavailableError(
            f"Manifest does not exist: {path}"
        ) from error

    except PermissionError as error:
        raise ArtifactUnavailableError(
            f"Permission denied while reading manifest: {path}"
        ) from error

    except UnicodeDecodeError as error:
        raise ArtifactContentError(
            f"Manifest is not valid UTF-8 text: {path}"
        ) from error

    except json.JSONDecodeError as error:
        raise ArtifactContentError(
            f"Manifest contains invalid JSON at line {error.lineno}: {path}"
        ) from error

    except OSError as error:
        raise ArtifactUnavailableError(
            f"Could not read manifest {path}: {error}"
        ) from error

    else:
        if not isinstance(raw, dict):
            raise ArtifactContentError(
                f"Expected a JSON object at the top level: {path}"
            )

        return raw

Several design decisions are doing real work here:

• FileNotFoundError and PermissionError are handled before their parent, OSError.
• UnicodeDecodeError and json.JSONDecodeError identify bad content, not a failed filesystem.
• The file object closes whether json.load() succeeds, fails to parse, or raises another exception, because it is inside with.
• The else block runs only when the try block completes successfully. It keeps post-read shape checking outside the protected I/O region.
• raise ... from error preserves the causal chain. A traceback will show both the application-facing error and the original operating-system or parsing error.

That last point is especially important for services. ArtifactUnavailableError is useful to code that decides whether to show a setup message, retry later, or fail a request. The chained FileNotFoundError remains useful to developers investigating the event.

After this function returns, validate the dictionary using the Pydantic model you designed in the prior lesson:

raw_manifest = load_manifest_json(Path("data/manifests/corpus_manifest.json"))
manifest = CorpusManifest.model_validate(raw_manifest)

A ValidationError here means something quite different from unreadable JSON: the artifact was successfully loaded but violated the expected schema. Do not hide that distinction by catching all errors in one large block.

A useful operational mapping looks like this:

Failure	Meaning	Typical application response
FileNotFoundError	Required local artifact is absent	Create it only if first-run behavior is intentional; otherwise fail with setup guidance
PermissionError	Process lacks filesystem rights	Report deployment/configuration fault; do not retry blindly
IsADirectoryError	Code expected a file but received a directory	Treat as an invalid path/configuration
UnicodeDecodeError	Bytes are not readable using the declared encoding	Reject the artifact as malformed
JSONDecodeError	Text is not valid JSON	Reject it and identify location where possible
ConnectionError	A connection could not be established or was interrupted	Often transient; surface a service-unavailable outcome
Timeout	Upstream did not respond in time	Bound latency and consider retry policy only where safe
HTTP 4xx / 5xx	Server responded, but status indicates failure	Classify based on status and product semantics

Notice that an HTTP 404 is not a connection failure: the network request completed and the server replied. It becomes an exception only when your client turns an unsuccessful status into one, typically with raise_for_status().

---

Network failures: timeouts, status checks, and targeted handling

A network call has more stages than a local file read:

1. Establish a connection.
2. Wait for a response.
3. Receive an HTTP response and evaluate its status.
4. Read and parse the response body.
5. Validate the resulting data against your schema.

Each stage can fail differently. A reliable client makes the stages explicit rather than assuming “a response object exists” means “the operation succeeded.”

Here is a small boundary function for an upstream model registry. It returns None for a missing card because this function’s hypothetical contract explicitly defines a 404 as an acceptable absence. In a different product, a missing card may be an error instead.

import json
from typing import Any

import requests


class UpstreamUnavailableError(Exception):
    """The upstream service could not be reached reliably."""


class UpstreamResponseError(Exception):
    """The upstream service returned an unusable response."""


def fetch_model_card(url: str) -> dict[str, Any] | None:
    try:
        with requests.get(url, timeout=10) as response:
            if response.status_code == 404:
                return None

            response.raise_for_status()
            body = response.content

    except requests.Timeout as error:
        raise UpstreamUnavailableError(
            f"Timed out while requesting model card"
        ) from error

    except requests.ConnectionError as error:
        raise UpstreamUnavailableError(
            f"Could not connect to model registry"
        ) from error

    except requests.HTTPError as error:
        raise UpstreamResponseError(
            f"Model registry returned HTTP {error.response.status_code}"
        ) from error

    except requests.RequestException as error:
        raise UpstreamUnavailableError(
            f"Unexpected request failure"
        ) from error

    try:
        payload: object = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise UpstreamResponseError(
            "Model registry returned malformed JSON"
        ) from error

    if not isinstance(payload, dict):
        raise UpstreamResponseError(
            "Model registry returned JSON that was not an object"
        )

    return payload

The order of the request handlers is deliberate:

• Timeout, ConnectionError, and HTTPError permit distinct product behavior and observability labels.
• RequestException is the final catch-all for the requests library’s remaining expected request failures.
• Parsing happens after the request block, so malformed JSON is not mislabeled as an unavailable network.
• A Pydantic model should validate payload before downstream application code uses it.

Always set a timeout. Without one, an unresponsive upstream can occupy a worker indefinitely, eventually exhausting concurrency or request capacity. Later in the course, you will add bounded concurrency, retries, backoff, and idempotency rules; for now, the essential behavior is to bound the request and classify its failure accurately.

Avoid placing credentials in URLs or echoing full URLs blindly in error messages and logs. Query strings can contain API keys, tokens, or user-provided content. Record useful operational context—service name, request ID, status code, elapsed time—without accidentally recording secrets.

---

Cleanup: prefer with; reserve finally for unconditional work

The try statement can include four clauses:

• try: code that may raise a handled exception.
• except: the matching recovery, translation, or fallback behavior.
• else: code that runs only if the try suite completed normally.
• finally: code that runs regardless of normal completion or an exception, barring abrupt process termination.

For resource lifetime, the default cleanup tool is a context manager:

with path.open("rb") as file:
    data = file.read()

with requests.get(url, timeout=10) as response:
    response.raise_for_status()
    body = response.content

In both cases, the resource is closed as the with block exits, even when its body raises. This is clearer and safer than initializing a variable to None, opening a resource manually, and attempting to close it later in finally.

Use finally when you have cleanup or reporting that is not already owned by a context manager. A classic artifact-writing example is removal of a partial download. You do not want a failed download to masquerade as a complete cached model file on the next run.

import logging
from pathlib import Path

import requests


logger = logging.getLogger(__name__)


def remove_if_present(path: Path) -> None:
    try:
        path.unlink(missing_ok=True)
    except OSError:
        # Preserve the original failure; cleanup itself is best effort.
        logger.exception("Could not remove partial artifact: %s", path)


def download_to_cache(url: str, destination: Path) -> None:
    partial_path = destination.with_suffix(destination.suffix + ".part")
    completed = False

    try:
        destination.parent.mkdir(parents=True, exist_ok=True)

        with requests.get(url, stream=True, timeout=30) as response:
            response.raise_for_status()

            with partial_path.open("wb") as file:
                for chunk in response.iter_content(chunk_size=64 * 1024):
                    if chunk:
                        file.write(chunk)

        partial_path.replace(destination)
        completed = True

    finally:
        if not completed:
            remove_if_present(partial_path)

This function has three cleanup layers:

1. The response context manager releases the network connection.
2. The file context manager closes and flushes the partial file.
3. The finally clause removes the partial artifact if the operation did not complete.

Two details are worth carrying into production code:

• Do not return from finally. It can suppress an exception or override a return value from the rest of the function.
• Do not let cleanup mask the original failure. remove_if_present() catches and logs its own OSError, allowing the connection, HTTP, or disk error that caused the incomplete download to remain the primary exception.

The fixed .part filename is adequate for a single-writer local script. Multiple concurrent workers need unique temporary names and coordination; otherwise, they can overwrite or delete each other’s partial artifacts.

---

Design failure behavior before writing handlers

Exception handling is easiest to reason about when each function states what callers can expect. For example:

def load_manifest_json(path: Path) -> dict[str, Any]:
    ...

Its operational contract may be:

• Returns a top-level JSON object on success.
• Raises ArtifactUnavailableError when it cannot access the path.
• Raises ArtifactContentError when the artifact cannot be decoded or parsed.
• Lets unexpected implementation errors propagate.

Likewise:

def fetch_model_card(url: str) -> dict[str, Any] | None:
    ...

Its contract may be:

• Returns a JSON object for a successful model-card response.
• Returns None only for an intentional 404 absence.
• Raises UpstreamUnavailableError for timeout or connectivity failures.
• Raises UpstreamResponseError for unsuccessful HTTP statuses or malformed response bodies.

This style pays off immediately in API handlers and background jobs. A caller can map ArtifactUnavailableError to a setup or deployment failure, distinguish it from a user-correctable missing resource, and record UpstreamUnavailableError separately from malformed upstream payloads. It also makes tests direct: each meaningful exception path is an explicit behavior rather than an accidental side effect of a generic catch block.

---

Key takeaways

Reliable Python code treats file and network failures as meaningful, typed outcomes.

• Catch the most specific exception types first, with broad parent classes such as OSError or RequestException last.
• Keep try blocks narrow so unrelated bugs are not accidentally caught.
• Distinguish unavailable resources, malformed representations, invalid schemas, and programming errors.
• Use raise NewError(...) from error when translating a low-level exception into an application-level contract.
• Call raise_for_status() so failed HTTP responses fail at the network boundary rather than corrupting later logic.
• Set explicit request timeouts.
• Prefer with for files and HTTP responses; use finally for unconditional cleanup or reporting that context managers do not already own.
• Never allow cleanup logic or a return in finally to hide the original error.

Next, you will test these contracts with pytest: fixtures will create temporary artifacts, parameterization will cover success and failure cases, and mocks will simulate network outcomes without making real upstream requests.