from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from enum import Enum
from pydantic import BaseModel, field_validator
_DATE_FORMATS = ("%Y-%m-%d", "%m/%d/%Y", "%d-%b-%Y", "%m/%d/%y")
class TransactionKind(str, Enum):
debit = "debit"
credit = "credit"
class Transaction(BaseModel):
posted_at: date
description: str
amount: Decimal
kind: TransactionKind
@field_validator("posted_at", mode="before")
@classmethod
def parse_date(cls, value):
if isinstance(value, date):
return value
text = str(value).strip()
for fmt in _DATE_FORMATS:
try:
return datetime.strptime(text, fmt).date()
except ValueError:
continue
raise ValueError(f"unrecognized date format: {text!r}")
@field_validator("amount", mode="before")
@classmethod
def parse_amount(cls, value):
if isinstance(value, Decimal):
return value
text = str(value).strip()
negative = text.startswith("(") and text.endswith(")")
cleaned = text.strip("()").replace("$", "").replace(",", "").strip()
try:
amount = Decimal(cleaned or "0")
except InvalidOperation as exc:
raise ValueError(f"invalid amount: {text!r}") from exc
return -amount if negative else amount
@field_validator("description")
@classmethod
def clean_description(cls, value):
return " ".join(value.split())
@field_validator("kind", mode="before")
@classmethod
def infer_kind(cls, value, info):
if value:
return str(value).strip().lower()
amount = info.data.get("amount", Decimal(0))
return TransactionKind.credit if amount >= 0 else TransactionKind.debit
import csv
import sys
from dataclasses import dataclass
from pathlib import Path
from pydantic import ValidationError
from transaction import Transaction
_HEADER_ALIASES = {
"date": "posted_at",
"posted": "posted_at",
"memo": "description",
"note": "description",
"type": "kind",
}
@dataclass
class RowError:
row: int
message: str
def _canonical_header(name):
key = name.strip().lower()
return _HEADER_ALIASES.get(key, key)
def read_transactions(path):
records = []
errors = []
with Path(path).open(newline="", encoding="utf-8-sig") as handle:
reader = csv.reader(handle)
raw_header = next(reader, None)
if raw_header is None:
return records, errors
header = [_canonical_header(col) for col in raw_header]
for line_no, row in enumerate(reader, start=2):
data = dict(zip(header, row))
try:
records.append(Transaction.model_validate(data))
except ValidationError as exc:
first = exc.errors()[0]
errors.append(RowError(line_no, first["msg"]))
return records, errors
if __name__ == "__main__":
txns, bad_rows = read_transactions(sys.argv[1])
print(f"accepted {len(txns)} rows, rejected {len(bad_rows)}")
for err in bad_rows:
print(f" row {err.row}: {err.message}")
This snippet shows a small ingestion pipeline that turns a messy bank-export CSV into clean, typed Transaction records. The core problem is that raw CSVs are stringly-typed and inconsistent: dates come in several formats, amounts carry currency symbols and thousands separators, and columns may be named differently between exports. Rather than scattering float() and strptime() calls across the codebase, the parsing and coercion logic is centralized in a single validated model.
In transaction.py, the Transaction model uses Pydantic's field_validator hooks to normalize input before it becomes a typed attribute. The amount field is stored as Decimal — never float — because binary floating point cannot represent money exactly, and rounding errors accumulate. parse_amount strips $, commas, and whitespace, and interprets parentheses as negative values, a common accounting convention. parse_date walks a list of accepted formats and raises a clear error if none match, so a single bad row surfaces a precise message instead of a cryptic stack trace. TransactionKind is an Enum so downstream code branches on a closed set of values rather than free-form strings.
In normalize.py, read_transactions handles the file-level concerns. _canonical_header lowercases and maps aliases (memo/note to description, date/posted to posted_at) so exports with different column names still parse. Each row is validated independently; a ValidationError is caught, recorded with its 1-based row number, and collected rather than aborting the whole file. This partial-success strategy matters for real imports, where operators want the good rows loaded and a report of the rejects. The function returns both the successful Transaction list and a list of RowError objects.
The if __name__ block in normalize.py demonstrates the intended usage: parse a file, then print a summary of accepted versus rejected rows. The trade-off of this design is a slightly heavier per-row cost from validation, which is negligible next to I/O for typical exports. Reaching for this pattern makes sense whenever untrusted tabular data must become trustworthy domain objects with a clear boundary between raw input and validated state.
Related snips
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler
from sklearn.linear_model import LogisticRegression
standard_pipeline = Pipeline([
('scaler', StandardScaler()),Scaling and normalization choices for different model families
class SignupForm
include ActiveModel::Model
include ActiveModel::Attributes
attribute :account_name, :string
attribute :email, :stringShallow Controller, Deep Params: Form Object Pattern
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Form Validation Example</title>
<style>HTML forms with validation and accessibility
module EmailNormalization
extend ActiveSupport::Concern
included do
attr_accessor :soft_warningsSoft Validation: Normalize + Validate Email
import axios from 'axios';
export type NormalizedErrors = {
fields: Record<string, string>;
formLevel: string | null;
};Frontend: normalize and display server validation errors
Share this code
Here's the card — post it anywhere.
