Modify partners

A modification workflow that modifies a partner.

initial_input_form_generator()

Gather input from the user needed for modifying a partner.

Source code in gso/workflows/tasks/modify_partners.py

def initial_input_form_generator() -> FormGenerator:
    """Gather input from the user needed for modifying a partner."""

    class SelectPartnerForm(FormPage):
        model_config = ConfigDict(title="Choose a Partner")

        partners: partner_choice()  # type: ignore[valid-type]

    initial_user_input = yield SelectPartnerForm

    partner = get_partner_by_name(name=initial_user_input.partners.name)

    class ModifyPartnerForm(SubmitFormPage):
        model_config = ConfigDict(title="Modify a Partner")

        name: str | None = partner.name
        email: EmailStr | None = partner.email

        @field_validator("name")
        def validate_name(cls, name: str) -> str | None:
            if partner.name == name:
                return None
            if filter_partners_by_name(name=name, case_sensitive=False):
                msg = "Partner with this name already exists."
                raise ValueError(msg)
            return name

        @field_validator("email")
        def validate_email(cls, email: str) -> EmailStr | None:
            if partner.email == email:
                return None
            if filter_partners_by_email(email=email, case_sensitive=False):
                msg = "Partner with this email already exists."
                raise ValueError(msg)
            return email

    user_input = yield ModifyPartnerForm

    return user_input.model_dump() | {"partner_id": partner.partner_id}

save_partner_to_database(partner_id, name, email)

Save modified partner in database.

Source code in gso/workflows/tasks/modify_partners.py

@step("Save partner information to database")
def save_partner_to_database(
    partner_id: UUIDstr,
    name: str | None,
    email: EmailStr | None,
) -> State:
    """Save modified partner in database."""
    partner = edit_partner(
        partner_data=ModifiedPartnerSchema(
            partner_id=partner_id,
            name=name,
            email=email,
        )
    )

    return {"modified_partner": partner}

task_modify_partners()

Modify a Partner.

Source code in gso/workflows/tasks/modify_partners.py

@workflow(
    "Modify partners",
    initial_input_form=initial_input_form_generator,
    target=Target.SYSTEM,
)
def task_modify_partners() -> StepList:
    """Modify a Partner."""
    return begin >> save_partner_to_database >> done