Http
This module contains several HTTP Steps that can be used to perform API Calls to HTTP endpoints
Example
from koheesio.steps.http import HttpGetStep
response = (
HttpGetStep(url="https://google.com").execute().json_payload
)
In the above example, the response
variable will contain the JSON response from the HTTP request.
koheesio.steps.http.HttpDeleteStep #
send DELETE requests
koheesio.steps.http.HttpGetStep #
send GET requests
Example
In the above example, theresponse
variable will contain the JSON response from the HTTP request.
koheesio.steps.http.HttpMethod #
Enumeration of allowed http methods
from_string
classmethod
#
Allows for getting the right Method Enum by simply passing a string value This method is not case-sensitive
koheesio.steps.http.HttpPostStep #
send POST requests
koheesio.steps.http.HttpPutStep #
send PUT requests
koheesio.steps.http.HttpStep #
Can be used to perform API Calls to HTTP endpoints
auth_header
class-attribute
instance-attribute
#
auth_header: Optional[SecretStr] = Field(
default=None,
description="[Optional] Authorization header",
alias="authorization_header",
examples=["Bearer <token>"],
)
data
class-attribute
instance-attribute
#
data: Union[Dict[str, str], str] = Field(
default_factory=dict,
description="[Optional] Data to be sent along with the request",
alias="body",
)
headers
class-attribute
instance-attribute
#
headers: Dict[str, Union[str, SecretStr]] = Field(
default={"Content-Type": "application/json"},
description="Request headers",
alias="header",
)
method
class-attribute
instance-attribute
#
method: Union[str, HttpMethod] = Field(
default=GET,
description="What type of Http call to perform. One of 'get', 'post', 'put', 'delete'. Defaults to 'get'.",
)
params
class-attribute
instance-attribute
#
params: Optional[Dict[str, Any]] = Field(
default_factory=dict,
description="[Optional] Set of extra parameters that should be passed to HTTP request",
)
session
class-attribute
instance-attribute
#
session: Session = Field(
default_factory=Session,
description="Existing requests session object to be used for making HTTP requests. If not provided, a new session object will be created.",
exclude=True,
repr=False,
)
timeout
class-attribute
instance-attribute
#
timeout: int = Field(
default=3, description="[Optional] Request timeout"
)
url
class-attribute
instance-attribute
#
url: str = Field(
default=..., description="API endpoint URL", alias="uri"
)
Output #
Output class for HttpStep
raw_payload
class-attribute
instance-attribute
#
raw_payload: Optional[str] = Field(
default=None,
alias="response_text",
description="The raw response for the request",
)
response_json
class-attribute
instance-attribute
#
response_json: Optional[Union[Dict, List]] = Field(
default=None,
alias="json_payload",
description="The JSON response for the request",
)
decode_sensitive_headers #
Authorization headers are being converted into SecretStr under the hood to avoid dumping any
sensitive content into logs by the encode_sensitive_headers
method.
However, when calling the get_headers
method, the SecretStr should be converted back to
string, otherwise sensitive info would have looked like '**********'.
This method decodes values of the headers
dictionary that are of type SecretStr into plain text.
Source code in src/koheesio/steps/http.py
delete #
encode_sensitive_headers #
encode_sensitive_headers() -> HttpStep
Encode potentially sensitive data into pydantic.SecretStr class to prevent them being displayed as plain text in logs.
Source code in src/koheesio/steps/http.py
execute #
Executes the HTTP request.
This method simply calls self.request()
, which includes the retry logic. If self.request()
raises an
exception, it will be propagated to the caller of this method.
Raises:
Type | Description |
---|---|
(RequestException, HTTPError)
|
The last exception that was caught if |
Source code in src/koheesio/steps/http.py
get #
get_proper_http_method_from_str_value #
Converts string value to HttpMethod enum value
Source code in src/koheesio/steps/http.py
post #
put #
set_outputs #
Types of response output
Source code in src/koheesio/steps/http.py
koheesio.steps.http.PaginatedHttpGetStep #
Represents a paginated HTTP GET step.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
paginate
|
bool
|
Whether to paginate the API response. Defaults to False. |
required |
pages
|
int
|
Number of pages to paginate. Defaults to 1. |
required |
offset
|
int
|
Offset for paginated API calls. Offset determines the starting page. Defaults to 1. |
required |
limit
|
int
|
Limit for paginated API calls. Defaults to 100. |
required |
limit
class-attribute
instance-attribute
#
limit: Optional[int] = Field(
default=100,
description="Limit for paginated API calls. The url should (optionally) contain a named limit parameter, for example: api.example.com/data?limit={limit}",
)
offset
class-attribute
instance-attribute
#
offset: Optional[int] = Field(
default=1,
description="Offset for paginated API calls. Offset determines the starting page. Defaults to 1. The url can (optionally) contain a named 'offset' parameter, for example: api.example.com/data?offset={offset}",
)
pages
class-attribute
instance-attribute
#
pages: Optional[int] = Field(
default=1,
description="Number of pages to paginate. Defaults to 1",
)
paginate
class-attribute
instance-attribute
#
paginate: Optional[bool] = Field(
default=False,
description="Whether to paginate the API response. Defaults to False. When set to True, the API response will be paginated. The url should contain a named 'page' parameter for example: api.example.com/data?page={page}",
)
execute #
Executes the HTTP GET request and handles pagination.
Returns:
Type | Description |
---|---|
Output
|
The output of the HTTP GET request. |
Source code in src/koheesio/steps/http.py
get_options #
get_options() -> dict
Returns the options to be passed to the requests.request() function.
Returns:
Type | Description |
---|---|
dict
|
The options. |