Skip to content

Step

Step models module.

Step

Bases: BaseSQLModel

Step model.

Attributes:

Name Type Description
name str

Step name.
pipeline Optional[Pipeline]

Pipeline.
executions List[StepExecution]

Step executions.
Source code in devpipe/models/step.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class Step(BaseSQLModel, table=True):
    """Step model.

    Attributes:
        name (str): Step name.
        pipeline (Optional[Pipeline]): Pipeline.
        executions (List[StepExecution]): Step executions.
    """

    __tablename__ = TABLES.step
    name: str = Field(max_length=128)
    pipeline: Optional["Pipeline"] = Relationship(
        back_populates="steps",
        link_model=PipelineStepLink,
    )
    executions: List["StepExecution"] = Relationship(
        back_populates="step",
    )

    @classmethod
    def get(
        cls,
        pipeline: "Pipeline",
        name: str,
        session: Session,
    ) -> "Pipeline":
        """Get or create step.

        Args:
            pipeline (Pipeline): Pipeline.
            name (str): Step name.
            session (Session): Database session.

        Returns:
            Step.
        """
        pipe = session.get(Pipeline, pipeline.uuid)
        statement = (
            select(cls)
            .join(PipelineStepLink)
            .join(Pipeline)
            .where(
                Pipeline.uuid == pipe.uuid,
                cls.name == name,
            )
        )
        step = session.exec(statement).first()
        if step:
            return step
        step = cls(name=name, pipeline=pipe)
        session.add(step)
        session.commit()
        session.refresh(step)
        return step

get(pipeline, name, session) classmethod

Get or create step.

Parameters:

Name Type Description Default
pipeline Pipeline

Pipeline.

 required
name str

Step name.

 required
session Session

Database session.

 required

Returns:

Type Description
Pipeline

Step.
Source code in devpipe/models/step.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def get(
    cls,
    pipeline: "Pipeline",
    name: str,
    session: Session,
) -> "Pipeline":
    """Get or create step.

    Args:
        pipeline (Pipeline): Pipeline.
        name (str): Step name.
        session (Session): Database session.

    Returns:
        Step.
    """
    pipe = session.get(Pipeline, pipeline.uuid)
    statement = (
        select(cls)
        .join(PipelineStepLink)
        .join(Pipeline)
        .where(
            Pipeline.uuid == pipe.uuid,
            cls.name == name,
        )
    )
    step = session.exec(statement).first()
    if step:
        return step
    step = cls(name=name, pipeline=pipe)
    session.add(step)
    session.commit()
    session.refresh(step)
    return step

StepExecution

Bases: BaseExecution

Step execution model.

Attributes:

Name Type Description
step_id Optional[str]

Step ID.
meta Dict[str, Any]

Metadata.
step Optional[Step]

Step.
pipeline_executions List[PipelineExecution]

Pipeline executions.
inputs Optional[StepInputs]

Step inputs.
outputs Optional[StepOutputs]

Step outputs.
Source code in devpipe/models/step.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class StepExecution(BaseExecution, table=True):
    """Step execution model.

    Attributes:
        step_id (Optional[str]): Step ID.
        meta (Dict[str, Any]): Metadata.
        step (Optional[Step]): Step.
        pipeline_executions (List[PipelineExecution]): Pipeline executions.
        inputs (Optional[StepInputs]): Step inputs.
        outputs (Optional[StepOutputs]): Step outputs.
    """

    __tablename__ = TABLES.step_exec
    step_id: Optional[str] = Field(
        default=None,
        foreign_key=f"{TABLES.step}.uuid",
    )
    meta: Dict[str, Any] = Field(
        default_factory=dict,
        sa_column=Column(JSON),
    )
    step: Optional["Step"] = Relationship(
        back_populates="executions",
    )
    pipeline_executions: List["PipelineExecution"] = Relationship(
        back_populates="step_executions",
        link_model=PipelineStepExecutionsLink,
    )
    inputs: Optional["StepInputs"] = Relationship(
        back_populates="step_execution",
    )
    outputs: Optional["StepOutputs"] = Relationship(
        back_populates="step_execution",
    )

    @classmethod
    def get(
        cls,
        step: "Step",
        inputs: "StepInputs",
        session: Session,
    ) -> Optional["StepExecution"]:
        """Get step execution from inputs.

        Args:
            step (Step): Step.
            inputs (StepInputs): Step inputs.
            session (Session): Database session.

        Returns:
            Step execution if found, otherwise None.
        """
        statement = (
            select(cls)
            .join(Step)
            .join(StepInputs)
            .where(
                Step.uuid == step.uuid,
                StepInputs.hash == inputs.hash,
            )
            .order_by(cls.created_at.desc())
        )
        return session.exec(statement).first()

get(step, inputs, session) classmethod

Get step execution from inputs.

Parameters:

Name Type Description Default
step Step

Step.

 required
inputs StepInputs

Step inputs.

 required
session Session

Database session.

 required

Returns:

Type Description
Optional[StepExecution]

Step execution if found, otherwise None.
Source code in devpipe/models/step.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@classmethod
def get(
    cls,
    step: "Step",
    inputs: "StepInputs",
    session: Session,
) -> Optional["StepExecution"]:
    """Get step execution from inputs.

    Args:
        step (Step): Step.
        inputs (StepInputs): Step inputs.
        session (Session): Database session.

    Returns:
        Step execution if found, otherwise None.
    """
    statement = (
        select(cls)
        .join(Step)
        .join(StepInputs)
        .where(
            Step.uuid == step.uuid,
            StepInputs.hash == inputs.hash,
        )
        .order_by(cls.created_at.desc())
    )
    return session.exec(statement).first()

StepInputs

Bases: BaseArtifact

Step inputs model.

Attributes:

Name Type Description
step_execution_id Optional[str]

Step execution ID.
step_execution Optional[StepExecution]

Step execution.
Source code in devpipe/models/step.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class StepInputs(BaseArtifact, table=True):
    """Step inputs model.

    Attributes:
        step_execution_id (Optional[str]): Step execution ID.
        step_execution (Optional[StepExecution]): Step execution.
    """

    __tablename__ = TABLES.step_inps
    step_execution_id: Optional[str] = Field(
        default=None,
        foreign_key=f"{TABLES.step_exec}.uuid",
    )
    step_execution: Optional["StepExecution"] = Relationship(
        back_populates="inputs",
    )

StepOutputs

Bases: BaseArtifact

Step outputs model.

Attributes:

Name Type Description
step_execution_id Optional[str]

Step execution ID.
step_execution Optional[StepExecution]

Step execution.
Source code in devpipe/models/step.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class StepOutputs(BaseArtifact, table=True):
    """Step outputs model.

    Attributes:
        step_execution_id (Optional[str]): Step execution ID.
        step_execution (Optional[StepExecution]): Step execution.
    """

    __tablename__ = TABLES.step_outs
    step_execution_id: Optional[str] = Field(
        default=None,
        foreign_key=f"{TABLES.step_exec}.uuid",
    )
    step_execution: Optional["StepExecution"] = Relationship(
        back_populates="outputs",
    )