from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List
@dataclass(frozen=True)
class Author:
id: int
name: str
email: str = ""
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Author":
return cls(
id=int(data["id"]),
name=data["name"].strip(),
email=data.get("email", ""),
)
@dataclass
class Comment:
id: int
body: str
created_at: datetime
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Comment":
return cls(
id=int(data["id"]),
body=data["body"],
created_at=datetime.fromisoformat(data["created_at"]),
)
@dataclass
class Article:
id: int
title: str
author: Author
comments: List[Comment] = field(default_factory=list)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Article":
return cls(
id=int(data["id"]),
title=data["title"],
author=Author.from_dict(data["author"]),
comments=[Comment.from_dict(c) for c in data.get("comments", [])],
)
from __future__ import annotations
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Union
from entities import Article, Author
@dataclass
class ImportError:
payload: Dict[str, Any]
reason: str
@dataclass
class ImportResult:
articles: List[Article] = field(default_factory=list)
errors: List[ImportError] = field(default_factory=list)
class FeedImporter:
@classmethod
def load(cls, source: Union[str, Path]) -> ImportResult:
raw = json.loads(Path(source).read_text(encoding="utf-8"))
return cls.from_records(raw.get("articles", []))
@classmethod
def from_records(cls, records: List[Dict[str, Any]]) -> ImportResult:
result = ImportResult()
for record in records:
try:
result.articles.append(Article.from_dict(record))
except (KeyError, ValueError, TypeError) as exc:
result.errors.append(ImportError(payload=record, reason=str(exc)))
return result
@staticmethod
def _dedupe_authors(articles: List[Article]) -> List[Author]:
by_id: Dict[int, Author] = {}
for article in articles:
by_id.setdefault(article.author.id, article.author)
return list(by_id.values())
from datetime import datetime
from entities import Comment
from feed_importer import FeedImporter
FIXTURE = [
{
"id": 1,
"title": "Normalizing feeds",
"author": {"id": "42", "name": " Ada "},
"comments": [
{"id": 9, "body": "nice", "created_at": "2024-01-02T10:00:00"},
],
},
{"id": 2, "author": {"id": 7, "name": "Grace"}}, # missing title
]
def test_nested_comments_are_typed():
result = FeedImporter.from_records(FIXTURE)
article = result.articles[0]
assert isinstance(article.comments[0], Comment)
assert isinstance(article.comments[0].created_at, datetime)
assert article.author.name == "Ada"
def test_bad_record_is_collected_not_raised():
result = FeedImporter.from_records(FIXTURE)
assert len(result.articles) == 1
assert len(result.errors) == 1
assert "title" in result.errors[0].reason
def test_authors_dedupe_by_id():
result = FeedImporter.from_records(FIXTURE)
authors = FeedImporter._dedupe_authors(result.articles)
assert len(authors) == 1
A common ETL problem is turning a deeply nested third-party JSON feed into flat, strongly-typed domain objects that the rest of an application can consume without reaching into raw dictionaries. This snippet shows how to do that with plain dataclasses and a small set of from_dict factory methods, keeping the mapping logic in one place and the rest of the code free of dict.get calls.
The entities.py tab defines the target shape. Each entity is a @dataclass — Author, Comment, and Article — and normalization means the nested comment array on the feed becomes a real list of Comment objects rather than nested dicts. The from_dict classmethods act as the boundary between untyped input and typed output: Author.from_dict coerces the id to int, Comment.from_dict parses the ISO timestamp into a datetime, and Article.from_dict fans out over the raw comments list to build child objects. Defaults such as field(default_factory=list) keep the type honest when the feed omits a key.
The feed_importer.py tab drives the traversal. FeedImporter.load reads the feed and returns a flat ImportResult that separates a list of articles from a list of errors, so one malformed record does not abort the whole batch. The importer catches KeyError and ValueError per article and records the offending payload, a pattern that matters when ingesting real feeds where a single missing field is routine. _dedupe_authors shows a second normalization step: authors are collapsed by id so the same person referenced across many articles is stored once.
The test_importer.py tab documents the contract with a fixture. It asserts that nested comments become Comment instances, that timestamps are parsed to datetime, and that a record missing title lands in errors rather than raising. This tests the resilience guarantee explicitly.
The trade-off is that hand-written from_dict methods are more verbose than a schema library like pydantic, but they add zero dependencies and make the coercion rules obvious. This approach fits well for small-to-medium feeds where the shape is known and clarity beats configuration.
Related snips
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
class PostSerializer < ActiveModel::Serializer
attributes :id, :title, :excerpt, :body, :published_at, :views, :likes_count, :comments_count
attribute :can_edit, if: :current_user_can_edit?
belongs_to :author, serializer: UserSummarySerializerSerializers with ActiveModel::Serializers
class PostsController < ApplicationController
def index
posts = Post.for_feed.page(params[:page]).per(25)
render json: {
data: posts.map { |post| PostSerializer.new(post).as_json },N+1 Proof Serialization with preloaded associations
Share this code
Here's the card — post it anywhere.
