1. Feature Request

A blog's reach scales with the volume and freshness of its content. Two existing external services — Medium and Reddit — already host posts that writers want to migrate. Give a username on the external service, pull their posts, normalise each one into an article on this blog, and report how many were imported, how many were skipped because they already exist, and how many failed our editorial policies.

This is an external capability again. The domain change stays small — one file inside the existing article aggregate. Infrastructure absorbs the scraping work, one adapter per source.

1.1 API Contract

I start, again, from the contract. The shape of /articles/imports decides the dispatcher pattern in HTTP and the use-case signature in application/.

Endpoint	Body	Success	Errors
`POST /articles/imports?source=<src>`	`{ account }`	`200 { source, account, imported, skipped, failed }`	`422` unknown source · `502` upstream format changed · `503` unreachable

source is a query parameter; values are medium, reddit
account is the username on that service
adding a third source — Substack, dev.to — should not require reshaping the article aggregate

1.2 UI Behaviour

Element	Location	Behaviour
Source picker + account input	"Import" dialog	Pick source, type account, hit import; disabled while empty
Progress indicator	Inside dialog	Spinner; on completion shows `{imported / skipped / failed}` counts
Inline error banner	Top of dialog	Shown on 502/503; the rest of the editor stays responsive

2. Implementation & Examples

Same walk: contract, then HTTP, then application, then domain, then the two infrastructure adapters. The domain abstraction is dictated by the use case: "iterate something that yields ArticleDraft for an account, persist what survives the policies, report counts." Everything else is plumbing.

2.1 HTTP Contracts

# src/http/contracts/imports.py

from pydantic import Field

from src.domain.articles import ExternalSource
from src.http._base import PublicModel


class ImportRequest(PublicModel):
    account: str = Field(min_length=1)


class ImportReportPublic(PublicModel):
    source: ExternalSource
    account: str
    imported: int
    skipped: int
    failed: int

2.2 HTTP Resources

The new dispatcher is the same query-param pattern Part 2 introduced for /actions. One route, match on ExternalSource, one concrete adapter per branch.

# src/http/resources/articles.py  (imports route shown; CRUD + actions omitted)

from src.domain.articles import ExternalSource
from src.http.contracts.imports import ImportReportPublic, ImportRequest
from src.infrastructure.integrations import (
    MediumArticleSource,
    RedditArticleSource,
)


@router.post("/imports", status_code=status.HTTP_200_OK)
@transactional
async def article_imports(
    body: ImportRequest,
    source: ExternalSource,
) -> ImportReportPublic:
    repository = SqlAlchemyArticlesRepository()

    match source:
        case ExternalSource.MEDIUM:
            adapter = MediumArticleSource()
        case ExternalSource.REDDIT:
            adapter = RedditArticleSource()

    report = await articles.import_account_articles(
        repository=repository,
        source=adapter,
        account=body.account,
    )
    return ImportReportPublic.model_validate(report)

FastAPI validates ?source= against ExternalSource — an unknown value is 422 before the handler runs
@transactional wraps the handler so every add_article call inside the use case commits together
min_length=1 on ImportRequest.account rejects empty strings at the contract level

2.3 Application Layer

One function. It iterates the source, applies the article aggregate's policies to every fetched draft (the same validate_policies() from Part 1), and persists through the existing BookshelfRepository.

# src/application/articles.py  (import_account_articles shown; others omitted)

from src.domain.articles import (
    BookshelfRepository,
    ExternalArticleSource,
    ImportReport,
)
from src.domain.errors import ArticleNotFound, DomainError


async def import_account_articles(
    repository: BookshelfRepository,
    source: ExternalArticleSource,
    account: str,
) -> ImportReport:
    imported = skipped = failed = 0

    async for draft in source.fetch(account):
        try:
            draft.validate_policies()
        except DomainError:
            failed += 1
            continue

        try:
            await repository.article(draft.slug)
        except ArticleNotFound:
            await repository.add_article(draft)
            imported += 1
        else:
            skipped += 1

    return ImportReport(
        source=source.kind,
        account=account,
        imported=imported,
        skipped=skipped,
        failed=failed,
    )

Three things to notice:

the use case knows nothing about RSS, HTTP, or Reddit's JSON shape — it works against ExternalArticleSource, the domain ABC
the Part 1 forbidden-words policy applies to inbound drafts the same way it applies to outbound writes and AI output — one rule, three callers
a slug collision is skipped, not an error; re-running the import on the same account is idempotent

2.4 Domain Layer. Inbound

An inbound import is "an article coming in from somewhere else", so the contract sits next to the rest of the article aggregate. One file, three pieces: the enum that names supported sources, the report shape, and the abstract contract any source must satisfy.

# src/domain/articles/inbound.py

import abc
from collections.abc import AsyncIterator
from enum import StrEnum

from src.domain.base import DomainModel

from .entities import ArticleDraft


class ExternalSource(StrEnum):
    MEDIUM = "medium"
    REDDIT = "reddit"


class ImportReport(DomainModel):
    source: ExternalSource
    account: str
    imported: int = 0
    skipped: int = 0
    failed: int = 0


class ExternalArticleSource(abc.ABC):
    # Subclasses set `kind` so the use case can stamp the report with
    # the source name without asking which subclass it received.
    kind: ExternalSource

    @abc.abstractmethod
    def fetch(self, account: str) -> AsyncIterator[ArticleDraft]:
        """Yield ArticleDrafts found under `account` on this source."""

fetch returns AsyncIterator[ArticleDraft] — the same ArticleDraft from Part 1. No new article shape was needed: an external post becomes an ArticleDraft as soon as it crosses the boundary, and everything downstream is what Part 1 already wrote.

src/domain/articles/__init__.py re-exports the three new names alongside the existing entities and the repository contract.

2.5 Domain Errors

# src/domain/errors/__init__.py  (additions only)

class ExternalSourceUnreachable(DomainError):
    def __init__(self, reason: str) -> None:
        super().__init__(f"External source unreachable: {reason}")
        self.reason = reason


class ExternalSourceFormatChanged(DomainError):
    def __init__(self, reason: str) -> None:
        super().__init__(f"External source format changed: {reason}")
        self.reason = reason

Unreachable → 503: the upstream did not respond (timeout, 5xx, DNS)
FormatChanged → 502: the upstream responded with a shape we did not expect (broken RSS, missing JSON key)

2.6 Infrastructure Layer. Medium Adapter

Medium publishes an RSS feed per account. The adapter fetches it, parses each <item> into an ArticleDraft, and translates httpx errors into domain errors before they escape.

# src/infrastructure/integrations/medium.py

from collections.abc import AsyncIterator
from datetime import date
from email.utils import parsedate_to_datetime
from xml.etree import ElementTree as ET

import httpx

from src.domain.articles import (
    ArticleDraft,
    ExternalArticleSource,
    ExternalSource,
)
from src.domain.errors import (
    ExternalSourceFormatChanged,
    ExternalSourceUnreachable,
)


class MediumArticleSource(ExternalArticleSource):
    kind: ExternalSource = ExternalSource.MEDIUM
    feed_url_template: str = "https://medium.com/feed/@{account}"
    request_timeout_seconds: float = 10.0

    async def fetch(self, account: str) -> AsyncIterator[ArticleDraft]:
        url = self.feed_url_template.format(account=account)

        try:
            async with httpx.AsyncClient(
                timeout=self.request_timeout_seconds,
            ) as client:
                response = await client.get(url)
                response.raise_for_status()
        except httpx.HTTPError as exc:
            raise ExternalSourceUnreachable(f"medium: {exc}") from exc

        try:
            root = ET.fromstring(response.text)
        except ET.ParseError as exc:
            raise ExternalSourceFormatChanged(f"medium: {exc}") from exc

        channel = root.find("channel")
        if channel is None:
            raise ExternalSourceFormatChanged("medium: feed missing <channel>")

        for item in channel.findall("item"):
            yield self._to_draft(item)

    def _to_draft(self, item: ET.Element) -> ArticleDraft:
        title: str = (item.findtext("title") or "").strip()
        link: str = (item.findtext("link") or "").strip()
        description: str = (item.findtext("description") or "").strip()
        pub_date_text: str | None = item.findtext("pubDate")

        if not title or not link:
            raise ExternalSourceFormatChanged(
                "medium: <item> missing title or link"
            )

        return ArticleDraft(
            title=title,
            slug=self._slug_from_url(link),
            summary=(description[:500] or title),
            body=description or title,
            published_on=self._parse_date(pub_date_text),
        )

    def _slug_from_url(self, url: str) -> str:
        return url.rstrip("/").rsplit("/", 1)[-1].split("?")[0]

    def _parse_date(self, text: str | None) -> date:
        if not text:
            return date.today()
        try:
            return parsedate_to_datetime(text).date()
        except (TypeError, ValueError):
            return date.today()

The feed URL template, the request timeout, the URL-to-slug helper, and the date parser all live on the class — they describe the Medium adapter, so they belong to its class rather than to module-level constants.

2.7 Infrastructure Layer. Reddit Adapter

Reddit exposes a user's posts as JSON at https://www.reddit.com/user/<account>.json. Same shape of adapter — fetch, parse, map provider exceptions into domain errors.

# src/infrastructure/integrations/reddit.py

from collections.abc import AsyncIterator
from datetime import datetime, timezone
from typing import Any

import httpx

from src.domain.articles import (
    ArticleDraft,
    ExternalArticleSource,
    ExternalSource,
)
from src.domain.errors import (
    ExternalSourceFormatChanged,
    ExternalSourceUnreachable,
)


class RedditArticleSource(ExternalArticleSource):
    kind: ExternalSource = ExternalSource.REDDIT
    user_url_template: str = "https://www.reddit.com/user/{account}.json"
    user_agent: str = (
        "rest-ddd-fastapi-blog/0.3 (https://github.com/parfeniukink)"
    )
    request_timeout_seconds: float = 10.0

    async def fetch(self, account: str) -> AsyncIterator[ArticleDraft]:
        url = self.user_url_template.format(account=account)

        try:
            async with httpx.AsyncClient(
                timeout=self.request_timeout_seconds,
                headers={"User-Agent": self.user_agent},
            ) as client:
                response = await client.get(url)
                response.raise_for_status()
                payload: Any = response.json()
        except httpx.HTTPError as exc:
            raise ExternalSourceUnreachable(f"reddit: {exc}") from exc
        except ValueError as exc:
            raise ExternalSourceFormatChanged(f"reddit: {exc}") from exc

        try:
            children: list[dict[str, Any]] = payload["data"]["children"]
        except (KeyError, TypeError) as exc:
            raise ExternalSourceFormatChanged(f"reddit: {exc}") from exc

        for child in children:
            if child.get("kind") != "t3":
                continue  # comments — skip; only submissions become articles
            yield self._to_draft(child.get("data", {}))

    def _to_draft(self, data: dict[str, Any]) -> ArticleDraft:
        title: str = (data.get("title") or "").strip()
        permalink: str = data.get("permalink") or ""
        post_id: str = (data.get("id") or "").strip()
        selftext: str = (data.get("selftext") or "").strip()
        created_utc: float = float(data.get("created_utc") or 0)

        if not title or not permalink or not post_id:
            raise ExternalSourceFormatChanged(
                "reddit: post missing title, permalink, or id"
            )

        return ArticleDraft(
            title=title,
            slug=self._slug_from_post(permalink=permalink, post_id=post_id),
            summary=(selftext[:500] or title),
            body=selftext or title,
            published_on=datetime.fromtimestamp(
                created_utc, tz=timezone.utc,
            ).date(),
        )

    def _slug_from_post(self, *, permalink: str, post_id: str) -> str:
        title_slug = permalink.rstrip("/").rsplit("/", 1)[-1]
        return f"{title_slug}-{post_id}"

Same shape as Medium. The slug includes Reddit's stable post id so repeated titles do not collide. Adding a third source — Substack, dev.to — follows the same template: one file, one class, the constants and helpers all live inside it.

2.8 Error Mapping

Two new entries; the route is unchanged.

# src/infrastructure/application/error_handlers.py  (additions only)

async def external_source_unreachable_handler(
    _: Request, exc: ExternalSourceUnreachable
) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        content={"detail": str(exc), "reason": exc.reason},
    )


async def external_source_format_changed_handler(
    _: Request, exc: ExternalSourceFormatChanged
) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_502_BAD_GATEWAY,
        content={"detail": str(exc), "reason": exc.reason},
    )


ERROR_HANDLERS = (
    # ...existing entries...
    (ExternalSourceUnreachable, external_source_unreachable_handler),
    (ExternalSourceFormatChanged, external_source_format_changed_handler),
)

2.9 Testing

The adapters make real HTTP calls in production. Tests do not — they use a FakeArticleSource that yields a canned list of drafts (or raises an injected error).

# src/tests/fakes/inbound.py

from collections.abc import AsyncIterator

from src.domain.articles import (
    ArticleDraft,
    ExternalArticleSource,
    ExternalSource,
)


class FakeArticleSource(ExternalArticleSource):
    def __init__(
        self,
        drafts: list[ArticleDraft] | None = None,
        kind: ExternalSource = ExternalSource.MEDIUM,
        error: Exception | None = None,
    ) -> None:
        self._drafts = drafts or []
        self.kind = kind
        self._error = error

    async def fetch(self, account: str) -> AsyncIterator[ArticleDraft]:
        if self._error is not None:
            raise self._error
        for draft in self._drafts:
            yield draft

# src/tests/unit/test_imports.py  (one test shown; full file in the repo)

@pytest.mark.asyncio
async def test_import_skips_drafts_whose_slug_already_exists(
    seeded_repository,
) -> None:
    source = FakeArticleSource(
        drafts=[_draft("first-post"), _draft("new-post")],
    )
    report = await articles_use_cases.import_account_articles(
        repository=seeded_repository,
        source=source,
        account="someone",
    )
    assert report.imported == 1
    assert report.skipped == 1
    assert report.failed == 0

The full file also covers a clean import, the policy-violation-as-failed case, and source errors propagating through the use case.

3. In Next Articles

Part 4 flips the picture. A new internal feature — a seven-state article publication lifecycle with role-aware transitions, a split publication pipeline (mechanical gates submission, cognitive advises the supervisor on review), and a deliberate persistence-shape vs HTTP-contract distinction for the reject_message field — forces the domain to grow. The dependency direction does not move; only the layer that absorbs the change does.

The cost of the next adapter is the only honest measure of an ABC.

Related Links