import abc
from typing import IO
class Exporter(abc.ABC):
name: str = ""
@abc.abstractmethod
def export(self, rows: list[dict], out: IO[str]) -> None:
"""Write rows to the given text stream."""
raise NotImplementedError
import logging
from importlib.metadata import entry_points
from typing import Callable, Type
from .base import Exporter
logger = logging.getLogger(__name__)
ENTRY_POINT_GROUP = "myapp.exporters"
class RegistryError(RuntimeError):
pass
class ExporterRegistry:
_exporters: dict[str, Type[Exporter]] = {}
_discovered: bool = False
@classmethod
def register(cls, name: str) -> Callable[[Type[Exporter]], Type[Exporter]]:
def decorator(target: Type[Exporter]) -> Type[Exporter]:
if not (isinstance(target, type) and issubclass(target, Exporter)):
raise RegistryError(f"{target!r} is not an Exporter subclass")
if name in cls._exporters:
raise RegistryError(f"exporter {name!r} is already registered")
target.name = name
cls._exporters[name] = target
return target
return decorator
@classmethod
def discover_entry_points(cls) -> None:
if cls._discovered:
return
for ep in entry_points(group=ENTRY_POINT_GROUP):
try:
plugin = ep.load()
except Exception:
logger.warning("failed to load exporter plugin %r", ep.name, exc_info=True)
continue
if ep.name not in cls._exporters:
cls.register(ep.name)(plugin)
cls._discovered = True
@classmethod
def get(cls, name: str) -> Type[Exporter]:
cls.discover_entry_points()
try:
return cls._exporters[name]
except KeyError:
available = ", ".join(sorted(cls._exporters)) or "none"
raise RegistryError(f"unknown exporter {name!r}; available: {available}")
@classmethod
def available(cls) -> list[str]:
cls.discover_entry_points()
return sorted(cls._exporters)
import csv
import json
from typing import IO
from .base import Exporter
from .registry import ExporterRegistry
@ExporterRegistry.register("json")
class JsonExporter(Exporter):
def export(self, rows: list[dict], out: IO[str]) -> None:
json.dump(rows, out, indent=2, default=str)
@ExporterRegistry.register("csv")
class CsvExporter(Exporter):
def export(self, rows: list[dict], out: IO[str]) -> None:
if not rows:
return
writer = csv.DictWriter(out, fieldnames=list(rows[0].keys()))
writer.writeheader()
writer.writerows(rows)
[project]
name = "myapp"
version = "1.2.0"
requires-python = ">=3.10"
[project.entry-points."myapp.exporters"]
# first-party plugins are found via import; downstream packages add lines like:
parquet = "myapp_parquet.exporter:ParquetExporter"
xlsx = "myapp_xlsx.exporter:XlsxExporter"
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
This snippet shows a small but complete plugin system for a data-export tool, built from a decorator-based in-process registry plus lazy discovery of third-party plugins through packaging entry points. The pattern solves a recurring extensibility problem: a host application needs to accept new implementations of some interface (exporters, codecs, transports) without hard-coding imports of every implementation, and ideally without knowing about plugins shipped in separate packages at all.
In registry.py, ExporterRegistry holds a dict mapping a string key to an Exporter subclass. The register classmethod returns a decorator so a class can be added at definition time with @ExporterRegistry.register("csv"). The decorator validates that the decorated class is actually a subclass of Exporter and rejects duplicate names, which turns silent overwrites into a loud RegistryError at import time. Registering the class object rather than an instance keeps construction lazy: the host decides when and how to instantiate, passing whatever constructor arguments a given call needs.
The interesting part is discover_entry_points, which pulls in plugins from other installed distributions. It uses importlib.metadata.entry_points to look up every entry point advertised under the myapp.exporters group. Each EntryPoint is only imported when ep.load() is called, so packages that expose plugins pay no import cost until discovery runs. A guard (_discovered) makes the method idempotent, and a try/except around ep.load() means one broken plugin package logs a warning instead of taking down the whole application. This is the same mechanism pytest, flake8, and many CLIs use for their plugin ecosystems.
The Exporter base class in base.py defines the contract: a name attribute and an export method that subclasses must implement. Making it an abc.ABC with an @abstractmethod means a half-finished plugin fails fast at instantiation rather than at the point of use.
plugins.py demonstrates first-party plugins. JsonExporter and CsvExporter register themselves simply by being imported, thanks to the decorator running at class-definition time. This is why the host must import the module for local plugins to appear, while external plugins arrive purely through entry points.
Finally, pyproject.toml shows the packaging side. The [project.entry-points."myapp.exporters"] table is where a downstream package would advertise parquet = mypkg.exporters:ParquetExporter; the string on the right is exactly what ep.load() imports and returns. The trade-offs are worth noting: entry-point discovery adds a small startup cost and depends on correct package metadata, and the loose string keys mean typos surface at runtime. In exchange the host stays completely decoupled from concrete implementations, and third parties extend it without editing its source. A developer reaches for this when building anything meant to be extended by plugins across package boundaries.
Related snips
import os
import stat
for root, _dirs, files in os.walk('/etc'):
for name in files:
path = os.path.join(root, name)Python security audit script for exposed risky filesystem state
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
from django.urls import path
from . import views
app_name = 'blog'
urlpatterns = [Django URL namespacing and reverse lookups
import graphene
from graphene_django import DjangoObjectType
from blog.models import Post, Comment
class PostType(DjangoObjectType):Django GraphQL with Graphene
from django.db.models import Count, Avg, Sum, Q, F
from django.views.generic import TemplateView
from products.models import Product, Order, OrderItem
class DashboardView(TemplateView):Django aggregation with annotate for statistics
from rest_framework import permissions
class IsOwner(permissions.BasePermission):
"""Allow only object owner to access."""Django REST Framework permissions and authorization
Share this code
Here's the card — post it anywhere.
