Skip to content

fastflowtransform.errors

FastFlowTransformError

Bases: Exception

Base class for all FastFlowTransform errors.

Attributes:

Name Type Description
message

Human-readable error message.
code

Optional short error code (e.g., 'CFG001', 'DAG002').
hint

Optional human hint with remediation steps.
Source code in src/fastflowtransform/errors.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class FastFlowTransformError(Exception):
    """
    Base class for all FastFlowTransform errors.

    Attributes:
        message: Human-readable error message.
        code: Optional short error code (e.g., 'CFG001', 'DAG002').
        hint: Optional human hint with remediation steps.
    """

    def __init__(self, message: str, *, code: str | None = None, hint: str | None = None):
        super().__init__(message)
        self.message = message
        self.code = code
        self.hint = hint

    def __str__(self) -> str:
        if self.hint:
            return f"{self.message}\n\nHint:\n{self.hint}"
        return self.message

DependencyNotFoundError

Bases: FastFlowTransformError

Raised when a model depends on another model that does not exist.

Source code in src/fastflowtransform/errors.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class DependencyNotFoundError(FastFlowTransformError):
    """Raised when a model depends on another model that does not exist."""

    def __init__(self, missing_map: dict[str, list[str]]):
        parts = []
        for depender, deps in missing_map.items():
            parts.append(f"{depender} → missing: {', '.join(sorted(deps))}")

        msg = (
            "❌ Missing model dependency.\n"
            + "\n".join(parts)
            + (
                "\n\nHints:\n"
                "• Check file names under models/ "
                "(node name = file stem, e.g. users.ff.sql → 'users.ff').\n"
                "• Ensure ref('…') matches the exact node name.\n"
                "• If it's a Python model, set @model(name='…')."
            )
        )

        super().__init__(msg)
        self.missing_map = missing_map

ModelCycleError

Bases: FastFlowTransformError

Raised when a cycle is detected in the model DAG.

Parameters:

Name Type Description Default
affected_nodes Iterable[str]

Nodes that couldn't be ordered due to the cycle.

 required
Source code in src/fastflowtransform/errors.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class ModelCycleError(FastFlowTransformError):
    """
    Raised when a cycle is detected in the model DAG.

    Args:
        affected_nodes: Nodes that couldn't be ordered due to the cycle.
    """

    def __init__(self, affected_nodes: Iterable[str]):
        affected = sorted(set(affected_nodes))
        msg = "Cycle detected in DAG. Affected models: " + ", ".join(affected)
        hint = (
            "Check for circular refs in your models:\n"
            "• Ensure A does not ref B while B (directly or indirectly) refs A.\n"
            "• Break the cycle by removing or refactoring one dependency.\n"
            "• If a ref is conditional in SQL, ensure the parse phase still sees the correct deps."
        )
        super().__init__(msg, code="DAG_CYCLE", hint=hint)
        self.affected_nodes = affected

ModuleLoadError

Bases: FastFlowTransformError

Raised when a Python model module cannot be loaded.

Source code in src/fastflowtransform/errors.py 
75
76
77
78
class ModuleLoadError(FastFlowTransformError):
    """Raised when a Python model module cannot be loaded."""

    pass

ProfileConfigError

Bases: FastFlowTransformError

Profile/configuration error with a short, actionable hint.

Source code in src/fastflowtransform/errors.py 
81
82
83
84
85
86
class ProfileConfigError(FastFlowTransformError):
    """Profile/configuration error with a short, actionable hint."""

    def __init__(self, message: str):
        # keep to a single line for CLI readability
        super().__init__(message.replace("\n", " ").strip())

ModelExecutionError

Bases: Exception

Raised when a model fails to execute/render on the engine. Carries friendly context for CLI formatting.

Source code in src/fastflowtransform/errors.py 
89
90
91
92
93
94
95
96
97
98
99
class ModelExecutionError(Exception):
    """Raised when a model fails to execute/render on the engine.
    Carries friendly context for CLI formatting.
    """

    def __init__(self, node_name: str, relation: str, message: str, sql_snippet: str | None = None):
        self.node_name = node_name
        self.relation = relation
        self.sql_snippet = sql_snippet
        self.message = message
        super().__init__(message)