Recommended resources
The freely available resources this lesson will direct you to.
Hello! Your previous script gave each recognizable input record a clear destination: accepted output or quarantine with a reason. It also printed batch counts at the end. But if the batch fails before that final line, how would you know which run failed, what it had processed, and whether the failure was bad input or a problem with the script?
In this lesson, you’ll add structured, contextual logging to that same local pipeline. The result will be one JSON log entry per event, with a run identifier and relevant fields. Allow about 40 minutes to study the short resources, make the change, and inspect a successful and a failed run. Python’s standard library is sufficient; there is nothing new to install.
Log events, not just messages
A print() can report a final count. Logging gives events a severity level and a consistent format, so you can distinguish routine progress from a rejected row or a failed batch. Start with this short demonstration.
Watch Python Logging - Tutorial by Tech With Tim for the logging levels and the basic idea of configuring their output. We will use the standard library shown in the video, but build JSON output rather than its plain-text format.
In logging levels, notice why INFO and DEBUG do not appear at the default WARNING threshold. Then watch the threshold and output formatting. Focus on what the level controls versus what the formatter controls; you need not reproduce the video's file-based setup.
The Python Logging HOWTO gives the distinctions behind those examples. In particular, a logger creates events, a handler sends them somewhere, and a formatter determines their representation.
Read the selected parts of Vinay Sajip’s Python Logging HOWTO to establish which events merit logging and how the logging components fit together.
Under “Basic Logging Tutorial” → “When to use logging,” read levels and uses, including the severity table. Then, under “Advanced Logging Tutorial,” read the four components. Stop there; configuration files and additional handlers are unnecessary for this script.
For this pipeline, the choices are fairly concrete:
INFO: a batch started or completed.WARNING: one row violated the input contract, but processing continued because it was quarantined.ERRORwith an exception traceback: the batch could not complete, such as when its header was wrong or a file operation failed.
A quarantined row is not a successful row, but neither is it necessarily a failed batch. That distinction is worth preserving in the logs.
Give each event useful context
A message such as bad row forces someone to guess which run, source, and record it means. A more useful event has separate fields: run_id, source, record_number, and reason. Separate fields can be inspected without extracting values from a sentence.
The terminal image illustrates why context matters: severity labels and fields make events easier to locate, while a traceback explains an exception. Its display is for human reading; our script will write one JSON object per log entry instead.
The Logging Cookbook describes two ideas we will combine: attaching shared context through a LoggerAdapter, and representing events in a machine-parseable form. Its JSON example uses a different implementation from ours, so study the purpose and mechanism, not code to paste into this project.
Logging Cookbook — Python 3.14.2 documentation
Read the Python Logging Cookbook for shared event context and the rationale for structured output. We will use its LoggerAdapter idea with a small custom JSON formatter suited to our pipeline.
Under “Adding contextual information to your logging output” → “Using LoggerAdapters to impart contextual information,” start at “Sometimes you want logging output to contain contextual information” and read adapter context. Stop before the discussion of overriding process(). Then find “Implementing structured logging.” Read the JSON approach, including the first StructuredMessage example. Stop before the custom-encoder example. Notice that its example combines a message with JSON data; our formatter will make the entire log entry one JSON object.
Context must also be safe. We will log the source file’s basename and the data-record number, not the raw row. The quarantine file already retains rejected data for review; copying entire records into logs would expose more data than an operator needs and could make logs very large. The record number has the same limitation as in the previous lesson: it counts CSV records produced by the reader, not necessarily physical lines in the file.
Add JSON logging to the validated pipeline
Work in the same project directory and Python environment as the previous lesson. Keep validated_orders.py unchanged and make a new entry-point script:
cp validated_orders.py logged_orders.py
Replace the contents of logged_orders.py with the following. It imports your existing validation function and paid_sale() transformation rather than redefining their rules.
import csv
import json
import logging
import sys
from datetime import datetime, timezone
from pathlib import Path
from uuid import uuid4
from transforms import paid_sale
from validated_orders import FIELDS, RowValidationError, validate_order
logger = logging.getLogger(__name__)
class JsonFormatter(logging.Formatter):
"""Format one log event as one JSON object."""
CONTEXT_FIELDS = (
"run_id",
"source",
"record_number",
"reason",
"read",
"valid",
"quarantined",
"paid_sales",
)
def format(self, record):
event = {
"timestamp": datetime.fromtimestamp(
record.created, timezone.utc
).isoformat(timespec="milliseconds"),
"level": record.levelname,
"logger": record.name,
"event": record.getMessage(),
}
for field in self.CONTEXT_FIELDS:
if hasattr(record, field):
event[field] = getattr(record, field)
if record.exc_info:
event["exception"] = self.formatException(record.exc_info)
return json.dumps(event, ensure_ascii=False)
def configure_logging():
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
def main(source_path):
configure_logging()
context = {
"run_id": uuid4().hex,
"source": source_path.name,
}
run_log = logging.LoggerAdapter(logger, context)
accepted_path = Path("data/paid_sales_logged.jsonl")
quarantine_path = Path("data/quarantined_orders_logged.jsonl")
total = valid = bad = sales = 0
run_log.info("batch_started")
try:
with source_path.open(newline="", encoding="utf-8") as source:
reader = csv.DictReader(
source, restkey="_extra_fields", strict=True
)
if reader.fieldnames != FIELDS:
raise ValueError(
"CSV header does not match the expected schema"
)
with accepted_path.open("w", encoding="utf-8") as accepted:
with quarantine_path.open(
"w", encoding="utf-8"
) as quarantined:
for record_number, row in enumerate(reader, start=1):
total += 1
try:
order = validate_order(row)
except RowValidationError as exc:
bad += 1
rejected = {
"record_number": record_number,
"reason": str(exc),
"raw": row,
}
quarantined.write(
json.dumps(rejected) + "\n"
)
row_log = logging.LoggerAdapter(
logger,
{
**context,
"record_number": record_number,
"reason": str(exc),
},
)
row_log.warning("row_quarantined")
else:
valid += 1
if order["paid"]:
sale = paid_sale(order)
accepted.write(json.dumps(sale) + "\n")
sales += 1
summary_log = logging.LoggerAdapter(
logger,
{
**context,
"read": total,
"valid": valid,
"quarantined": bad,
"paid_sales": sales,
},
)
summary_log.info("batch_completed")
return 0
except Exception:
run_log.exception("batch_failed")
return 1
if __name__ == "__main__":
source = (
Path(sys.argv[1])
if len(sys.argv) > 1
else Path("data/orders_validation_demo.csv")
)
sys.exit(main(source))
Notice the division of responsibilities. LoggerAdapter supplies context shared by an event, such as the run ID. JsonFormatter takes the resulting log record and produces JSON. StreamHandler sends that JSON to the terminal’s standard error stream, keeping it separate from the accepted and quarantined data files.
The formatter selects known context fields rather than dumping every attribute of the internal log record. It writes timestamps in UTC and serializes an exception traceback as a JSON string. That traceback may contain escaped newline characters, but it remains one physical log line, so a JSON-lines reader can still parse the file.
There is one important boundary in the code. The inner except RowValidationError is still narrow: it quarantines a bad row and logs a warning. The outer handler records an unexpected batch failure with run_log.exception(), which includes the active traceback, and returns a non-zero exit status. It does not relabel a failed file write or programming error as malformed source data.
Inspect a completed run and a failed run
Run the script against the six-record fixture from the previous lesson:
python logged_orders.py 2> data/orders_run.log
The terminal will be quiet because the logs went to standard error and were redirected to data/orders_run.log. Inspect that file with cat data/orders_run.log. You should find a batch_started event, four row_quarantined warnings, and a batch_completed event. Their run_id values should match within this run. The completion event should contain read: 6, valid: 2, quarantined: 4, and paid_sales: 1.
You can verify that each line is parseable JSON with Python:
python -c 'import json; from pathlib import Path; events = [json.loads(line) for line in Path("data/orders_run.log").read_text().splitlines()]; print([(event["event"], event["level"]) for event in events])'
The two output data files have the same roles as before, under their new *_logged.jsonl names. Logs describe what happened; they are not a replacement for accepted data or quarantine records.
Now give the script a file with an incorrect header:
printf 'wrong,product,quantity,amount_cents,paid,coupon_code\n' > data/orders_bad_header_demo.csv
python logged_orders.py data/orders_bad_header_demo.csv 2> data/orders_failed.log
cat data/orders_failed.log
This run should log batch_started followed by batch_failed. The failure event has an exception field containing the ValueError traceback; it should have no batch_completed event. Because the header is checked before either output file is opened, this particular failure does not overwrite the earlier output files. A failure later in processing could leave partial output, as in the previous lesson; logging makes that failure visible but does not itself make output publication safe.
For a tiny fixture, one warning per rejected row is helpful. On a large feed, logging every bad row could overwhelm the useful events. You would still retain quarantine records, but might reduce row-level log volume while keeping the run summary and enough context to investigate failures.
Conclusion
Your pipeline now emits events with distinct severity levels, a shared run ID, safe record context, and machine-parseable JSON. A completed batch reports counts; a rejected row identifies its reason without exposing the raw record in the log; and a failed batch records a traceback and exits unsuccessfully. The key principle is to keep data handling, event reporting, and exception handling as separate responsibilities.
Next, you’ll write pytest tests for the transformation and validation functions so their expected behavior can be checked automatically as the project changes.