app

Example Flask application with RESTful API using flask_restx_marshmallow.

CreateTaskParameters dataclass

CreateTaskParameters(
    *, name: str, settings: dict | None = None
)

Parameters for creating a task.

DeleteTaskParameters dataclass

DeleteTaskParameters(*, task_id: UUID)

Parameters for deleting a task.

Model

Bases: _FSAModel

Add original type for type annotation, which is able to use typevar.

QueryTaskParameters dataclass

QueryTaskParameters(
    *, task_id: UUID | None = None, name: str | None = None
)

Parameters for querying a task.

QueryTasksParameters dataclass

QueryTasksParameters(*, name: str | None = None)

Parameters for querying a task.

SQLAlchemy

Bases: SQLAlchemy

Add original type for type annotation, which is able to use typevar.

StandardSchema

StandardSchema(
    message: str | None = None,
    header_fields: Iterable[str] | None = None,
    **kwargs: Unpack[SchemaInitKwargs],
)

Bases: make_default_schema(name='StandardResponse')

Standard response schema.

Set default message.

Parameters:

Name	Type	Description	Default
message	str | None	response message. Defaults to None.	None
header_fields	Iterable[str] | None	header fields. Defaults to None.	None
kwargs	Unpack[SchemaInitKwargs]	other kwargs.	{}

Source code in app.py

def __init__(
    self,
    message: str | None = None,
    header_fields: Iterable[str] | None = None,
    **kwargs: Unpack[SchemaInitKwargs],
) -> None:
    """Set default message.

    Args:
        message (str | None): response message. Defaults to None.
        header_fields (Iterable[str] | None): header fields. Defaults to None.
        kwargs (Unpack[SchemaInitKwargs]): other kwargs.
    """
    super().__init__(header_fields=header_fields, **kwargs)
    if message is not None:
        self.fields["message"].dump_default = message
        self.fields["message"].metadata = dict(
            **self.fields["message"].metadata,
        )
        self.fields["message"].metadata["example"] = message

Task

Task(
    *,
    task_id: UUID,
    name: str,
    settings: dict | None = None,
)

Bases: Model

Task model for managing tasks in the database.

Add init annotations for type checking.

Parameters:

Name	Type	Description	Default
task_id	UUID	The unique identifier for the task.	required
name	str	The name of the task.	required
settings	dict	The settings for the task. Defaults to None.	None

Source code in app.py

def __init__(
    self,
    *,
    task_id: UUID,
    name: str,
    settings: dict | None = None,
) -> None:
    """Add init annotations for type checking.

    Args:
        task_id (UUID): The unique identifier for the task.
        name (str): The name of the task.
        settings (dict, optional): The settings for the task.
            Defaults to None.
    """

create classmethod

create(
    create_task_info: CreateTaskParameters,
) -> TupleResponse[Self] | TupleResponse

Create a new task instance.

Parameters:

Name	Type	Description	Default
create_task_info	CreateTaskParameters	The parameters for creating.	required

Returns:

Name	Type	Description
TupleResponse	TupleResponse[Self] | TupleResponse	A response containing the created task or an error. message.

Source code in app.py

@classmethod
def create(
    cls,
    create_task_info: "CreateTaskParameters",
) -> TupleResponse["Self"] | TupleResponse:
    """Create a new task instance.

    Args:
        create_task_info (CreateTaskParameters): The parameters for creating.

    Returns:
        TupleResponse: A response containing the created task or an error.
            message.
    """
    try:
        task_ = cls(
            task_id=uuid4(),
            name=create_task_info.name,
            settings=create_task_info.settings,
        )
        db.session.add(task_)
        db.session.commit()
        return TupleResponse(
            Success(
                message="create task successfully",
                result=Result(
                    items=[task_],
                    total=1,
                ),
            ),
        )
    except Exception:
        db.session.rollback()
        db.session.commit()
        logger.exception("task already exists: %s", create_task_info.name)
        return TupleResponse(
            Warn(message="task already exists"),
            status_code=HTTPStatus.BAD_REQUEST,
        )

delete classmethod

delete(task_id: UUID) -> TupleResponse

Delete a task instance by its ID.

Parameters:

Name	Type	Description	Default
task_id	UUID	The unique identifier for the task.	required

Returns:

Name	Type	Description
TupleResponse	TupleResponse	A response indicating the success or failure of the deletion.

Source code in app.py

@classmethod
def delete(
    cls,
    task_id: UUID,
) -> TupleResponse:
    """Delete a task instance by its ID.

    Args:
        task_id (UUID): The unique identifier for the task.

    Returns:
        TupleResponse: A response indicating the success or failure of the
            deletion.
    """
    if (query := cls.query.filter_by(task_id=task_id)).one_or_none():
        query.delete()
        return TupleResponse(Success(message="delete task successfully"))
    logger.warning("task not found: %s", task_id)
    return TupleResponse(
        Warn(message="task not found"),
        status_code=HTTPStatus.BAD_REQUEST,
    )

query_task classmethod

query_task(
    params: QueryTaskParameters,
) -> TupleResponse[Self]

Query a task instance by its ID or name.

Parameters:

Name	Type	Description	Default
params	QueryTaskParameters	The parameters for querying the task.	required

Returns:

Name	Type	Description
TupleResponse	TupleResponse[Self]	A response containing the queried task or an error message.

Source code in app.py

@classmethod
def query_task(cls, params: "QueryTaskParameters") -> TupleResponse["Self"]:
    """Query a task instance by its ID or name.

    Args:
        params (QueryTaskParameters): The parameters for querying the task.

    Returns:
        TupleResponse: A response containing the queried task or an error
            message.
    """
    if task := cls.query.filter_by(task_id=params.task_id).one_or_none():
        return TupleResponse(
            Success(
                message="query task successfully",
                result=Result(items=[task], total=1),
            ),
        )
    logger.warning("task not found: %s", params)
    return TupleResponse(
        Warn(message="task not found"),
        status_code=HTTPStatus.BAD_REQUEST,
    )

query_tasks classmethod

query_tasks(
    params: QueryTasksParameters,
) -> TupleResponse[Self]

Query a task instance by its ID or name.

Parameters:

Name	Type	Description	Default
params	QueryTasksParameters	The parameters for querying the task.	required

Returns:

Name	Type	Description
TupleResponse	TupleResponse[Self]	A response containing the queried task or an error message.

Source code in app.py

@classmethod
def query_tasks(cls, params: "QueryTasksParameters") -> TupleResponse["Self"]:
    """Query a task instance by its ID or name.

    Args:
        params (QueryTasksParameters): The parameters for querying the task.

    Returns:
        TupleResponse: A response containing the queried task or an error
            message.
    """
    query = cls.query
    if params.name is not None:
        query = query.filter(cls.name.ilike(params.name))
    if query.count() > 0:
        return TupleResponse(
            Success(
                message="query task successfully",
                result=Result(
                    items=query.all(),
                    total=query.count(),
                ),
            ),
        )
    logger.warning("task not found: %s", params)
    return TupleResponse(
        Warn(message="task not found"),
        status_code=HTTPStatus.BAD_REQUEST,
    )

update classmethod

update(
    task_update_info: UpdateTaskParameters,
) -> TupleResponse

Update a task instance by its ID.

Parameters:

Name	Type	Description	Default
task_update_info	UpdateTaskParameters	The parameters for updating the task.	required

Returns:

Name	Type	Description
TupleResponse	TupleResponse	A response indicating the success or failure of the update.

Source code in app.py

@classmethod
def update(cls, task_update_info: "UpdateTaskParameters") -> TupleResponse:
    """Update a task instance by its ID.

    Args:
        task_update_info (UpdateTaskParameters): The parameters for updating
            the task.

    Returns:
        TupleResponse: A response indicating the success or failure of the
            update.
    """
    if task := cls.query.filter_by(
        task_id=task_update_info.task_id,
    ).one_or_none():
        if task_update_info.name is not None:
            task.name = task_update_info.name
        if task_update_info.settings is not None:
            task.settings = task_update_info.settings
        db.session.commit()
        return TupleResponse(Success(message="update task successfully"))
    logger.warning("task not found: %s", task_update_info.task_id)
    return TupleResponse(
        Warn(message="task not found"),
        status_code=HTTPStatus.BAD_REQUEST,
    )

TaskManage

Bases: Resource

task manage.

delete

delete(params: DeleteTaskParameters, **_) -> TupleResponse

Delete a task by its ID.

Parameters:

Name	Type	Description	Default
params	DeleteTaskParameters	The parameters for deleting the task.	required

Returns:

Name	Type	Description
TupleResponse	TupleResponse	A response indicating the success or failure of the deletion.

Source code in app.py

@ns.parameters(params=DeleteTaskParameters, location="path")
@ns.responses(message="delete task successfully")
@ns.responses(message="task not found", code=HTTPStatus.BAD_REQUEST)
def delete(self, params: "DeleteTaskParameters", **_) -> "TupleResponse":  # noqa: ANN003
    """Delete a task by its ID.

    Args:
        params (DeleteTaskParameters): The parameters for deleting the task.

    Returns:
        TupleResponse: A response indicating the success or failure of the
            deletion.
    """
    return Task.delete(params.task_id)

get

get(
    params: QueryTaskParameters, **_
) -> TupleResponse[Task]

Query a task by its ID or name.

Parameters:

Name	Type	Description	Default
params	QueryTaskParameters	The parameters for querying the task.	required

Returns:

Type	Description
TupleResponse[Task]	TupleResponse[Task]: A response containing the queried task.

Source code in app.py

@ns.parameters(params=QueryTaskParameters, location="query")
@ns.responses(schema=TaskSchema(message="query task success"))
@ns.responses(message="task not found", code=HTTPStatus.BAD_REQUEST)
def get(self, params: "QueryTaskParameters", **_) -> "TupleResponse[Task]":  # noqa: ANN003
    """Query a task by its ID or name.

    Args:
        params (QueryTaskParameters): The parameters for querying the task.

    Returns:
        TupleResponse[Task]: A response containing the queried task.
    """
    return Task.query_task(params)

patch

patch(params: UpdateTaskParameters, **_) -> TupleResponse

Update a task by its ID.

Parameters:

Name	Type	Description	Default
params	UpdateTaskParameters	The parameters for updating the task.	required

Returns:

Name	Type	Description
TupleResponse	TupleResponse	A response indicating the success or failure of the update.

Source code in app.py

@ns.parameters(params=UpdateTaskParameters, location="body")
@ns.responses(message="update task successfully")
@ns.responses(message="task not found", code=HTTPStatus.BAD_REQUEST)
def patch(self, params: "UpdateTaskParameters", **_) -> "TupleResponse":  # noqa: ANN003
    """Update a task by its ID.

    Args:
        params (UpdateTaskParameters): The parameters for updating the task.

    Returns:
        TupleResponse: A response indicating the success or failure of the
            update.
    """
    return Task.update(params)

TaskProfile dataclass

TaskProfile(
    *,
    task_id: UUID,
    name: str,
    settings: dict | None = None,
)

Task profile schema.

TaskProfileSchema

Bases: class_schema(TaskProfile, base=Schema)

Task profile schema for marshmallow serialization.

TaskSchema

TaskSchema(
    message: str | None = None,
    header_fields: Iterable[str] | None = None,
    **kwargs: Unpack[SchemaInitKwargs],
)

Bases: StandardSchema

Query response schema for task.

Set default message.

Parameters:

Name	Type	Description	Default
message	str | None	response message. Defaults to None.	None
header_fields	Iterable[str] | None	header fields. Defaults to None.	None
kwargs	Unpack[SchemaInitKwargs]	other kwargs.	{}

Source code in app.py

def __init__(
    self,
    message: str | None = None,
    header_fields: Iterable[str] | None = None,
    **kwargs: Unpack[SchemaInitKwargs],
) -> None:
    """Set default message.

    Args:
        message (str | None): response message. Defaults to None.
        header_fields (Iterable[str] | None): header fields. Defaults to None.
        kwargs (Unpack[SchemaInitKwargs]): other kwargs.
    """
    super().__init__(header_fields=header_fields, **kwargs)
    if message is not None:
        self.fields["message"].dump_default = message
        self.fields["message"].metadata = dict(
            **self.fields["message"].metadata,
        )
        self.fields["message"].metadata["example"] = message

TasksManage

Bases: Resource

tasks manage.

get

get(params: QueryTasksParameters) -> TupleResponse[Task]

Query all tasks.

Parameters:

Name	Type	Description	Default
params	QueryTasksParameters	The parameters for querying tasks.	required

Returns:

Type	Description
TupleResponse[Task]	TupleResponse[Task]: A response containing the queried tasks.

Source code in app.py

@ns.parameters(params=QueryTasksParameters, location="query")
@ns.responses(schema=TaskSchema(message="query task success"))
@ns.responses(message="task not found", code=HTTPStatus.BAD_REQUEST)
def get(self, params: "QueryTasksParameters") -> "TupleResponse[Task]":
    """Query all tasks.

    Args:
        params (QueryTasksParameters): The parameters for querying tasks.

    Returns:
        TupleResponse[Task]: A response containing the queried tasks.
    """
    return Task.query_tasks(params)

post

post(
    params: CreateTaskParameters,
) -> TupleResponse[Task] | TupleResponse

Create a new task.

Parameters:

Name	Type	Description	Default
params	CreateTaskParameters	The parameters for creating a task.	required

Returns:

Type	Description
TupleResponse[Task] | TupleResponse	TupleResponse[Task]: A response containing the created task.

Source code in app.py

@ns.parameters(params=CreateTaskParameters, location="body")
@ns.responses(schema=TaskSchema(message="create task successfully"))
@ns.responses(message="task already exists", code=HTTPStatus.BAD_REQUEST)
def post(
    self,
    params: "CreateTaskParameters",
) -> "TupleResponse[Task] | TupleResponse":
    """Create a new task.

    Args:
        params (CreateTaskParameters): The parameters for creating a task.

    Returns:
        TupleResponse[Task]: A response containing the created task.
    """
    return Task.create(params)

UpdateTaskParameters dataclass

UpdateTaskParameters(
    *,
    task_id: UUID,
    name: str | None = None,
    settings: dict | None = None,
)

Parameters for updating a task.