StreamFlow Code Style Skill

Conventions for writing Python code in StreamFlow that are not automatically enforced by the formatter. For type annotations, load the StreamFlow Mypy Type Checking skill.

Run auto-fix before committing:

uv run make format codespell pyupgrade

Exclude streamflow/cwl/antlr from all checks. Use American English in all code, docstrings, and comments.

Imports: Two Manual Rules

Everything else is handled by ruff automatically. Two things ruff never does for you:

1. Always add from __future__ import annotations as the first import in every file:

from __future__ import annotations

2. Use TYPE_CHECKING to break circular imports:

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from streamflow.core.data import DataManager

Naming Conventions

Kind	Convention	Example
Classes	PascalCase	WorkflowExecutor
Functions / methods	snake_case	run_workflow
Constants	UPPER_SNAKE_CASE	MAX_RETRIES
Private members	_prefix	_internal_state
Type variables	Short uppercase	_KT, _VT, _T

Error Handling

Use StreamFlow's custom exceptions — never raise bare Exception or RuntimeError.

from streamflow.core.exception import WorkflowExecutionException
from streamflow.log_handler import logger

try:
    result = await process()
except SpecificException as e:
    logger.exception(e)
    raise WorkflowExecutionException(f"Failed to process: {e}") from e

Available exceptions (from streamflow.core.exception):

Exception	When to use
WorkflowException	General workflow errors
WorkflowDefinitionException	Invalid workflow definition
WorkflowExecutionException	Runtime execution failures
WorkflowProvenanceException	Provenance/tracking errors
FailureHandlingException	Fault tolerance failures
InvalidPluginException	Plugin loading/validation errors
ProcessorTypeError	Type mismatches in processors

Async Cleanup Pattern

Use asyncio.gather with create_task for concurrent teardown, with a finally block to guarantee database closure:

async def close(self) -> None:
    try:
        await asyncio.gather(
            asyncio.create_task(self.manager.close()),
            asyncio.create_task(self.scheduler.close()),
        )
    except Exception as e:
        logger.exception(e)
    finally:
        await self.database.close()

Docstrings

Use Sphinx-style field lists. Every non-trivial public function/method should have a docstring.

def deploy_connector(
    self, name: str, config: ConnectorConfig, location: ExecutionLocation
) -> Connector:
    """
    Deploy a connector at the given execution location.

    :param name: Unique name for the connector instance
    :param config: Deployment configuration for the connector
    :param location: Target execution location
    :returns: The deployed connector instance
    :raises WorkflowExecutionException: If deployment fails
    """

• One-line summary on the first line, no blank line before it
• Blank line before :param block if a body paragraph follows
• Use :returns: (not :return:)
• Only document :raises: for exceptions callers should handle
• Do not repeat the type in the param description — types live in annotations

See Also

• StreamFlow Mypy Type Checking skill — type annotations and forbidden types
• StreamFlow Git Workflow skill — commit message format