Skip to content

Sphinx Style API Documentation

Documentation Style

This page demonstrates API documentation using Sphinx-style docstrings, which provide rich reStructuredText markup capabilities for complex documentation needs.

Overview

Sphinx-style docstrings use reStructuredText field lists with :param: and :return: tags. They're particularly well-suited for:

  • Complex enterprise projects
  • Libraries with extensive documentation
  • Projects using Sphinx documentation generator
  • APIs requiring rich formatting

DataProcessor Class

src.docstring_examples.sphinx_style.DataProcessor

DataProcessor(name: str, validation_enabled: bool = True, max_transformations: int = 100)

Comprehensive data processor with loading, transformation, and export.

This class provides a complete data processing pipeline including data loading, transformation operations, validation, and export functionality. It supports various data formats and provides extensive configuration options.

The processor maintains internal state and provides detailed logging of all operations for debugging and monitoring purposes.

:param name: Descriptive name for this processor instance :type name: str :param validation_enabled: Whether to enable data validation :type validation_enabled: bool, optional :param max_transformations: Maximum number of transformations allowed :type max_transformations: int, optional

.. note:: This processor is thread-safe for read operations but not for concurrent modifications. Use appropriate locking mechanisms if sharing across threads.

.. warning:: The processor has a maximum transformation limit to prevent infinite loops or excessive memory usage.

Examples::

# Create processor with validation enabled
processor = DataProcessor("sales_data", validation_enabled=True)

# Load data from various sources
processor.load_data({"product": "Widget", "sales": 1000})
processor.load_from_file("additional_data.json")

# Apply transformations
processor.transform_data(
    lambda x: x * 1.1 if isinstance(x, (int, float)) else x
)
processor.apply_filter(lambda item: item.get("sales", 0) > 500)

# Export results
processor.export_data("processed_results.json")

.. versionadded:: 1.0.0 Initial implementation with basic processing capabilities

Initialize the data processor.

:param name: Descriptive name for this processor instance :type name: str :param validation_enabled: Whether to enable data validation :type validation_enabled: bool, optional :param max_transformations: Maximum number of transformations allowed :type max_transformations: int, optional :raises ValueError: If name is empty or max_transformations is negative

Examples::

# Basic processor
processor = DataProcessor("basic_processor")

# Advanced processor with custom settings
processor = DataProcessor(
    name="advanced_processor",
    validation_enabled=True,
    max_transformations=50
)
Source code in src/docstring_examples/sphinx_style.py
def __init__(
    self, name: str, validation_enabled: bool = True, max_transformations: int = 100
) -> None:
    """Initialize the data processor.

    :param name: Descriptive name for this processor instance
    :type name: str
    :param validation_enabled: Whether to enable data validation
    :type validation_enabled: bool, optional
    :param max_transformations: Maximum number of transformations allowed
    :type max_transformations: int, optional
    :raises ValueError: If name is empty or max_transformations is negative

    Examples::

        # Basic processor
        processor = DataProcessor("basic_processor")

        # Advanced processor with custom settings
        processor = DataProcessor(
            name="advanced_processor",
            validation_enabled=True,
            max_transformations=50
        )
    """
    super().__init__(name)

    if max_transformations < 0:
        raise ValueError("max_transformations must be non-negative")

    self.data: Optional[Dict[str, Any]] = None
    self.transformations_applied = 0
    self.export_count = 0
    self.validation_enabled = validation_enabled
    self.max_transformations = max_transformations

    logger.info(
        f"DataProcessor '{name}' initialized with "
        f"validation={'on' if validation_enabled else 'off'}"
    )

Attributes

status property

status: str

Get the current processor status.

:returns: String indicating current status ("active" or "inactive") :rtype: str

.. versionadded:: 1.0.0 Added status property for monitoring processor state

Functions

load_data

load_data(data: Union[Dict[str, Any], List[Dict[str, Any]]]) -> None

Load data into the processor.

This method accepts data in dictionary or list format and stores it internally for subsequent processing operations. The data is validated if validation is enabled.

:param data: Data to load - either a single dictionary or list of dictionaries :type data: dict or list of dict :raises ProcessingError: If data validation fails or processor is inactive :raises TypeError: If data is not in expected format

Examples::

# Load single record
processor.load_data({"id": 1, "name": "Alice", "score": 95})

# Load multiple records
processor.load_data([
    {"id": 1, "name": "Alice", "score": 95},
    {"id": 2, "name": "Bob", "score": 87}
])

.. note:: Data validation is performed if :attr:validation_enabled is True.

.. seealso:: :meth:load_from_file for loading data from files

Source code in src/docstring_examples/sphinx_style.py
def load_data(self, data: Union[Dict[str, Any], List[Dict[str, Any]]]) -> None:
    """Load data into the processor.

    This method accepts data in dictionary or list format and stores it
    internally for subsequent processing operations. The data is validated
    if validation is enabled.

    :param data: Data to load - either a single dictionary or list of dictionaries
    :type data: dict or list of dict
    :raises ProcessingError: If data validation fails or processor is inactive
    :raises TypeError: If data is not in expected format

    Examples::

        # Load single record
        processor.load_data({"id": 1, "name": "Alice", "score": 95})

        # Load multiple records
        processor.load_data([
            {"id": 1, "name": "Alice", "score": 95},
            {"id": 2, "name": "Bob", "score": 87}
        ])

    .. note::
       Data validation is performed if :attr:`validation_enabled` is True.

    .. seealso::
       :meth:`load_from_file` for loading data from files
    """
    if not self.is_active:
        raise ProcessingError("Cannot load data: processor is inactive")

    if not isinstance(data, (dict, list)):
        raise TypeError("Data must be a dictionary or list of dictionaries")

    if self.validation_enabled:
        self._validate_data(data)

    if isinstance(data, dict):
        self.data = {"records": [data]}
    else:
        self.data = {"records": data}

    logger.info(
        f"Loaded {len(self.data['records'])} record(s) into processor '{self.name}'"
    )

load_from_file

load_from_file(file_path: Union[str, Path]) -> None

Load data from a JSON file.

Reads data from the specified file path and loads it into the processor. Supports both string paths and Path objects.

:param file_path: Path to the JSON file to load :type file_path: str or pathlib.Path :raises ProcessingError: If file cannot be read or contains invalid JSON :raises FileNotFoundError: If the specified file does not exist :raises PermissionError: If insufficient permissions to read the file

Examples::

# Load from string path
processor.load_from_file("data/input.json")

# Load from Path object
from pathlib import Path
processor.load_from_file(Path("data") / "input.json")

.. note:: This method internally calls :meth:load_data after reading the file.

.. versionadded:: 1.0.0 Added file loading capability

Source code in src/docstring_examples/sphinx_style.py
def load_from_file(self, file_path: Union[str, Path]) -> None:
    """Load data from a JSON file.

    Reads data from the specified file path and loads it into the processor.
    Supports both string paths and Path objects.

    :param file_path: Path to the JSON file to load
    :type file_path: str or pathlib.Path
    :raises ProcessingError: If file cannot be read or contains invalid JSON
    :raises FileNotFoundError: If the specified file does not exist
    :raises PermissionError: If insufficient permissions to read the file

    Examples::

        # Load from string path
        processor.load_from_file("data/input.json")

        # Load from Path object
        from pathlib import Path
        processor.load_from_file(Path("data") / "input.json")

    .. note::
       This method internally calls :meth:`load_data` after reading the file.

    .. versionadded:: 1.0.0
       Added file loading capability
    """
    if not self.is_active:
        raise ProcessingError("Cannot load from file: processor is inactive")

    file_path = Path(file_path)

    try:
        with file_path.open("r", encoding="utf-8") as f:
            data = json.load(f)
        self.load_data(data)
        logger.info(f"Successfully loaded data from {file_path}")

    except FileNotFoundError:
        raise ProcessingError(f"File not found: {file_path}")
    except json.JSONDecodeError as e:
        raise ProcessingError(f"Invalid JSON in file {file_path}: {e}")
    except Exception as e:
        raise ProcessingError(
            f"Error loading file {file_path}: {e}", original_error=e
        )

transform_data

transform_data(transformation_func: Callable[[Any], Any]) -> ProcessingResult

Apply a transformation function to all data values.

Applies the provided transformation function to each value in the loaded data. The transformation preserves the data structure while modifying individual values.

:param transformation_func: Function to apply to each data value. Should accept any value and return the transformed value. :type transformation_func: callable :returns: Dictionary containing transformation results :rtype: dict :raises ProcessingError: If no data is loaded, processor is inactive, or max transformations exceeded :raises ValueError: If transformation_func is not callable

The returned dictionary contains the following keys:

  • records_processed (int): Number of records processed
  • transformations_applied (int): Total transformations applied to this dataset
  • success (bool): Whether the transformation completed successfully

Examples::

# Convert all strings to uppercase
result = processor.transform_data(
    lambda x: x.upper() if isinstance(x, str) else x
)

# Apply mathematical transformation to numbers
result = processor.transform_data(
    lambda x: x * 1.1 if isinstance(x, (int, float)) else x
)

# Complex transformation with type checking
def complex_transform(value):
    if isinstance(value, str):
        return value.strip().title()
    elif isinstance(value, (int, float)):
        return round(value * 1.05, 2)
    return value

result = processor.transform_data(complex_transform)

.. warning:: The processor enforces a maximum number of transformations to prevent infinite loops or excessive resource usage.

.. versionadded:: 1.0.0 Initial transformation capability

Source code in src/docstring_examples/sphinx_style.py
def transform_data(
    self, transformation_func: Callable[[Any], Any]
) -> ProcessingResult:
    """Apply a transformation function to all data values.

    Applies the provided transformation function to each value in the loaded data.
    The transformation preserves the data structure while modifying
    individual values.

    :param transformation_func: Function to apply to each data value.
        Should accept any value and return the transformed value.
    :type transformation_func: callable
    :returns: Dictionary containing transformation results
    :rtype: dict
    :raises ProcessingError: If no data is loaded, processor is
        inactive, or max transformations exceeded
    :raises ValueError: If transformation_func is not callable

    The returned dictionary contains the following keys:

    - ``records_processed`` (*int*): Number of records processed
    - ``transformations_applied`` (*int*): Total transformations
      applied to this dataset
    - ``success`` (*bool*): Whether the transformation completed successfully

    Examples::

        # Convert all strings to uppercase
        result = processor.transform_data(
            lambda x: x.upper() if isinstance(x, str) else x
        )

        # Apply mathematical transformation to numbers
        result = processor.transform_data(
            lambda x: x * 1.1 if isinstance(x, (int, float)) else x
        )

        # Complex transformation with type checking
        def complex_transform(value):
            if isinstance(value, str):
                return value.strip().title()
            elif isinstance(value, (int, float)):
                return round(value * 1.05, 2)
            return value

        result = processor.transform_data(complex_transform)

    .. warning::
       The processor enforces a maximum number of transformations to prevent
       infinite loops or excessive resource usage.

    .. versionadded:: 1.0.0
       Initial transformation capability
    """
    if not self.is_active:
        raise ProcessingError("Cannot transform data: processor is inactive")

    if self.data is None:
        raise ProcessingError("No data loaded for transformation")

    if not callable(transformation_func):
        raise ValueError("transformation_func must be callable")

    if self.transformations_applied >= self.max_transformations:
        raise ProcessingError(
            f"Maximum transformations ({self.max_transformations}) exceeded"
        )

    try:
        records_processed = 0
        for record in self.data["records"]:
            for key, value in record.items():
                record[key] = transformation_func(value)
            records_processed += 1

        self.transformations_applied += 1

        result = {
            "records_processed": records_processed,
            "transformations_applied": self.transformations_applied,
            "success": True,
        }

        logger.info(
            f"Applied transformation to {records_processed} records in "
            f"processor '{self.name}'"
        )
        return result

    except Exception as e:
        raise ProcessingError(f"Transformation failed: {e}", original_error=e)

apply_filter

apply_filter(filter_func: Callable[[Dict[str, Any]], bool]) -> ProcessingResult

Filter data records based on a predicate function.

Removes records that don't match the filter criteria. The filter function should return True for records to keep and False for records to remove.

:param filter_func: Predicate function that accepts a record dictionary and returns True to keep the record, False to remove it :type filter_func: callable :returns: Dictionary containing filter results :rtype: dict :raises ProcessingError: If no data is loaded or processor is inactive :raises ValueError: If filter_func is not callable

The returned dictionary contains the following keys:

  • records_before (int): Number of records before filtering
  • records_after (int): Number of records after filtering
  • records_removed (int): Number of records removed
  • success (bool): Whether the filter operation completed successfully

Examples::

# Keep only records with score > 80
result = processor.apply_filter(lambda record: record.get('score', 0) > 80)

# Keep records with specific status
result = processor.apply_filter(
    lambda record: record.get('status') == 'active'
)

# Complex filter with multiple conditions
def complex_filter(record):
    return (record.get('score', 0) > 70 and
           record.get('active', False) and
           len(record.get('name', '')) > 0)

result = processor.apply_filter(complex_filter)

.. note:: Filtering modifies the internal data structure by removing records that don't match the criteria.

.. versionadded:: 1.0.0 Added filtering capability

Source code in src/docstring_examples/sphinx_style.py
def apply_filter(
    self, filter_func: Callable[[Dict[str, Any]], bool]
) -> ProcessingResult:
    """Filter data records based on a predicate function.

    Removes records that don't match the filter criteria. The filter function
    should return True for records to keep and False for records to remove.

    :param filter_func: Predicate function that accepts a record
        dictionary and returns True to keep the record, False to remove it
    :type filter_func: callable
    :returns: Dictionary containing filter results
    :rtype: dict
    :raises ProcessingError: If no data is loaded or processor is inactive
    :raises ValueError: If filter_func is not callable

    The returned dictionary contains the following keys:

    - ``records_before`` (*int*): Number of records before filtering
    - ``records_after`` (*int*): Number of records after filtering
    - ``records_removed`` (*int*): Number of records removed
    - ``success`` (*bool*): Whether the filter operation completed successfully

    Examples::

        # Keep only records with score > 80
        result = processor.apply_filter(lambda record: record.get('score', 0) > 80)

        # Keep records with specific status
        result = processor.apply_filter(
            lambda record: record.get('status') == 'active'
        )

        # Complex filter with multiple conditions
        def complex_filter(record):
            return (record.get('score', 0) > 70 and
                   record.get('active', False) and
                   len(record.get('name', '')) > 0)

        result = processor.apply_filter(complex_filter)

    .. note::
       Filtering modifies the internal data structure by removing records
       that don't match the criteria.

    .. versionadded:: 1.0.0
       Added filtering capability
    """
    if not self.is_active:
        raise ProcessingError("Cannot apply filter: processor is inactive")

    if self.data is None:
        raise ProcessingError("No data loaded for filtering")

    if not callable(filter_func):
        raise ValueError("filter_func must be callable")

    try:
        records_before = len(self.data["records"])

        filtered_records = []
        for record in self.data["records"]:
            if filter_func(record):
                filtered_records.append(record)

        self.data["records"] = filtered_records
        records_after = len(filtered_records)
        records_removed = records_before - records_after

        result = {
            "records_before": records_before,
            "records_after": records_after,
            "records_removed": records_removed,
            "success": True,
        }

        logger.info(
            f"Filter applied: {records_removed} records removed, "
            f"{records_after} remaining"
        )
        return result

    except Exception as e:
        raise ProcessingError(f"Filter operation failed: {e}", original_error=e)

export_data

export_data(file_path: Union[str, Path], format: str = 'json') -> None

Export processed data to a file.

Saves the current processed data to the specified file path in the requested format. Currently supports JSON format with plans for additional formats in future versions.

:param file_path: Output file path for the exported data :type file_path: str or pathlib.Path :param format: Export format ("json" currently supported) :type format: str, optional :raises ProcessingError: If no data to export, processor inactive, or export fails :raises ValueError: If format is not supported :raises PermissionError: If insufficient permissions to write to the file

Examples::

# Basic JSON export
processor.export_data("output.json")

# Export with explicit format
processor.export_data("output.json", format="json")

# Export to Path object
from pathlib import Path
output_path = Path("exports") / "processed_data.json"
processor.export_data(output_path)

.. note:: The export operation creates parent directories if they don't exist.

.. todo:: Add support for CSV, XML, and other export formats.

.. versionadded:: 1.0.0 Initial JSON export capability

Source code in src/docstring_examples/sphinx_style.py
def export_data(self, file_path: Union[str, Path], format: str = "json") -> None:
    """Export processed data to a file.

    Saves the current processed data to the specified file path in the
    requested format. Currently supports JSON format with plans for
    additional formats in future versions.

    :param file_path: Output file path for the exported data
    :type file_path: str or pathlib.Path
    :param format: Export format ("json" currently supported)
    :type format: str, optional
    :raises ProcessingError: If no data to export, processor inactive,
        or export fails
    :raises ValueError: If format is not supported
    :raises PermissionError: If insufficient permissions to write to the file

    Examples::

        # Basic JSON export
        processor.export_data("output.json")

        # Export with explicit format
        processor.export_data("output.json", format="json")

        # Export to Path object
        from pathlib import Path
        output_path = Path("exports") / "processed_data.json"
        processor.export_data(output_path)

    .. note::
       The export operation creates parent directories if they don't exist.

    .. todo::
       Add support for CSV, XML, and other export formats.

    .. versionadded:: 1.0.0
       Initial JSON export capability
    """
    if not self.is_active:
        raise ProcessingError("Cannot export data: processor is inactive")

    if self.data is None:
        raise ProcessingError("No data to export")

    if format.lower() != "json":
        raise ValueError(f"Unsupported export format: {format}")

    file_path = Path(file_path)

    try:
        # Ensure parent directory exists
        file_path.parent.mkdir(parents=True, exist_ok=True)

        with file_path.open("w", encoding="utf-8") as f:
            json.dump(self.data, f, indent=2, ensure_ascii=False)

        self.export_count += 1
        logger.info(f"Exported data to {file_path} (export #{self.export_count})")

    except PermissionError:
        raise ProcessingError(f"Permission denied writing to {file_path}")
    except Exception as e:
        raise ProcessingError(f"Export failed: {e}", original_error=e)

get_statistics

get_statistics() -> Dict[str, Any]

Get comprehensive statistics about the processor and its data.

Returns detailed information about the current state of the processor, including data counts, transformation history, and processing metrics.

:returns: Dictionary containing comprehensive processor statistics :rtype: dict

The returned dictionary contains the following keys:

  • processor_name (str): Name of this processor instance
  • processor_status (str): Current status (active/inactive)
  • data_loaded (bool): Whether data is currently loaded
  • record_count (int): Number of records currently loaded
  • transformations_applied (int): Number of transformations applied
  • export_count (int): Number of times data has been exported
  • validation_enabled (bool): Whether validation is enabled
  • created_at (str): When the processor was created (ISO format)
  • uptime_seconds (float): How long the processor has existed in seconds

Examples::

stats = processor.get_statistics()
print(f"Processor: {stats['processor_name']}")
print(f"Records: {stats['record_count']}")
print(f"Transformations: {stats['transformations_applied']}")

.. note:: The uptime is calculated from processor creation time to current time.

.. versionadded:: 1.0.0 Added comprehensive statistics reporting

Source code in src/docstring_examples/sphinx_style.py
def get_statistics(self) -> Dict[str, Any]:
    """Get comprehensive statistics about the processor and its data.

    Returns detailed information about the current state of the processor,
    including data counts, transformation history, and processing metrics.

    :returns: Dictionary containing comprehensive processor statistics
    :rtype: dict

    The returned dictionary contains the following keys:

    - ``processor_name`` (*str*): Name of this processor instance
    - ``processor_status`` (*str*): Current status (active/inactive)
    - ``data_loaded`` (*bool*): Whether data is currently loaded
    - ``record_count`` (*int*): Number of records currently loaded
    - ``transformations_applied`` (*int*): Number of transformations applied
    - ``export_count`` (*int*): Number of times data has been exported
    - ``validation_enabled`` (*bool*): Whether validation is enabled
    - ``created_at`` (*str*): When the processor was created (ISO format)
    - ``uptime_seconds`` (*float*): How long the processor has existed in seconds

    Examples::

        stats = processor.get_statistics()
        print(f"Processor: {stats['processor_name']}")
        print(f"Records: {stats['record_count']}")
        print(f"Transformations: {stats['transformations_applied']}")

    .. note::
       The uptime is calculated from processor creation time to current time.

    .. versionadded:: 1.0.0
       Added comprehensive statistics reporting
    """
    current_time = datetime.now()
    uptime = (current_time - self.created_at).total_seconds()

    stats = {
        "processor_name": self.name,
        "processor_status": self.status,
        "data_loaded": self.data is not None,
        "record_count": len(self.data["records"]) if self.data else 0,
        "transformations_applied": self.transformations_applied,
        "export_count": self.export_count,
        "validation_enabled": self.validation_enabled,
        "created_at": self.created_at.isoformat(),
        "uptime_seconds": round(uptime, 2),
    }

    return stats

process

process(data: Any) -> Any

Process data using the internal pipeline.

Implementation of the abstract process method from BaseProcessor. This method provides a simplified interface for basic data processing.

:param data: Data to process :type data: Any :returns: Processed data :rtype: Any :raises ProcessingError: If processing fails

.. note:: This is a simplified interface. For advanced processing, use the specific methods like :meth:load_data, :meth:transform_data, etc.

.. versionadded:: 1.0.0 Implementation of abstract process method

Source code in src/docstring_examples/sphinx_style.py
def process(self, data: Any) -> Any:
    """Process data using the internal pipeline.

    Implementation of the abstract process method from BaseProcessor.
    This method provides a simplified interface for basic data processing.

    :param data: Data to process
    :type data: Any
    :returns: Processed data
    :rtype: Any
    :raises ProcessingError: If processing fails

    .. note::
       This is a simplified interface. For advanced processing, use the
       specific methods like :meth:`load_data`, :meth:`transform_data`, etc.

    .. versionadded:: 1.0.0
       Implementation of abstract process method
    """
    try:
        self.load_data(data)
        return self.data
    except Exception as e:
        raise ProcessingError(f"Processing failed: {e}", original_error=e)

deactivate

deactivate() -> None

Deactivate the processor.

Once deactivated, the processor should not perform any operations until reactivated.

.. note:: This method logs the deactivation event for monitoring purposes.

.. versionadded:: 1.0.0 Initial implementation of processor deactivation

Source code in src/docstring_examples/sphinx_style.py
def deactivate(self) -> None:
    """Deactivate the processor.

    Once deactivated, the processor should not perform any operations
    until reactivated.

    .. note::
       This method logs the deactivation event for monitoring purposes.

    .. versionadded:: 1.0.0
       Initial implementation of processor deactivation
    """
    self.is_active = False
    logger.info(f"Processor '{self.name}' deactivated")

ProcessingError Exception

src.docstring_examples.sphinx_style.ProcessingError

ProcessingError(message: str, error_code: Optional[str] = None, original_error: Optional[Exception] = None)

Bases: Exception

Custom exception for data processing errors.

This exception is raised when data processing operations fail due to invalid data, configuration errors, or runtime issues.

:param message: Error message describing the failure :type message: str :param error_code: Optional error code for categorization :type error_code: str or None :param original_error: Original exception that caused this error :type original_error: Exception or None

.. versionadded:: 1.0.0 Initial implementation of ProcessingError

Initialize ProcessingError.

:param message: Descriptive error message :type message: str :param error_code: Optional categorization code :type error_code: str or None, optional :param original_error: The original exception if this is a wrapper :type original_error: Exception or None, optional

Module-Level Functions

Functions Coming Soon

Module-level function documentation will be added when the source code is available.

Example Usage

from docstring_examples.sphinx_style import DataProcessor

# Create a processor instance
processor = DataProcessor(
    name="document_processor",
    validation_enabled=True,
    max_transformations=20
)

# Load and process document metadata
processor.load_data(document_data)
processor.transform_data(format_academic_text)
processor.apply_filter(substantial_docs_filter)

# Export processed metadata
processor.export_data('documents.json')

Style Benefits

Rich Markup

  • Full reStructuredText support
  • Cross-references and links
  • Math notation support
  • Complex formatting options

Documentation Features

  • Type annotations with :type: tags
  • Detailed parameter descriptions
  • Exception documentation
  • See Also sections

Best Practices

  • Use consistent field names
  • Document all parameters and returns
  • Include type information
  • Leverage reStructuredText features