Fluent API

Given

Represents the Given stage of the request where you can specify parameters, headers, cookies and the body of the request.

Parameters:

Name	Type	Description	Default
client	Client	The client instance to use for making the request.	required

query_param(params)

Adds a query parameter to the request.

Parameters:

Name	Type	Description	Default
params	Dict[str, Any]	A dictionary of query parameters to add.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> params = {'chocolate': 'chip'}
>>> r = given(client).query_param(params).when("GET", "/cookies/set").then()...

Returns:

Name	Type	Description
Given	Given	The instance of the Given class.

cookies(cookies)

Adds multiple cookies to the request.

Parameters:

Name	Type	Description	Default
cookies	Dict[str, Any]	A dictionary of cookies to add.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> cks = {"cookie1": "value1", "cookie2": "value2"}
>>> given(client).cookies(cks).when("GET", "/cookies").then()...

Returns:

Name	Type	Description
Given	Given	The instance of the Given class.

header(key, value)

Adds a header to the request.

Parameters:

Name	Type	Description	Default
key	str	The key of the header.	required
value	Any	The value of the header.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).header("Authorization", "Bearer TOKEN").when("GET", "/headers").then()...

Returns:

Name	Type	Description
Given	Given	The instance of the Given class.

headers(headers)

Adds multiple headers to the request.

Parameters:

Name	Type	Description	Default
headers	Dict[str, Any]	A dictionary of headers to add.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> HEADERS = {'Authorization': 'Bearer TOKEN', 'test_header': 'test_value'}
>>> given(client).headers(HEADERS).                  when("GET", "/headers").                  then()...

Returns:

Name	Type	Description
Given	Given	The instance of the Given class.

body(content=None, *, json=None, data=None)

Sets the body of the request. Either json or data can be set, but not both.

Parameters:

Name	Type	Description	Default
content	dict	Shortcut for setting JSON data directly. Defaults to None.	None
json	Any	The JSON body to set for the request. Defaults to None.	None
data	Any	The form data to send in the body of the request. Defaults to None.	None

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> # Using `content` as a shortcut for JSON data
>>> given(client).body({"key": "value"}).when("POST", "/post").then()...
>>> # Explicitly using `json` parameter
>>> given(client).body(json={"key": "value"}).when("POST", "/post").then()...
>>> # Using `data` for form data
>>> given(client).body(data="key=value").when("POST", "/post").then()...

Raises:

Type	Description
ValueError	If both json and data are provided.

Returns:

Name	Type	Description
Given	Given	The instance of the Given class.

file_upload(field_name, file_path)

Sets the file to upload for the request. Args: field_name (str): The name of the form field the file is associated with. file_path (str): The path to the file to be uploaded.

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).file_upload("userfile", "data/test.png").            >>>     when("POST", "/doc/file_upload.html").then()...

Note

field_name must be the same as the name of the form field in the request.

Returns:

Name	Type	Description
Given	Given	The instance of the Given class for chaining.

when(method=None, url='')

Transitions from the Given stage to the When stage, where the request is made.

Parameters:

Name	Type	Description	Default
method	str	The HTTP method to use.	None
url	str	The URL to send the request to.	''

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/get").then()...

Note

If the url is not provided, the url provided in the Client instance will be used.

Returns:

Name	Type	Description
When	When	The instance of the When class.

When

Represents the When stage of the request where the actual request is made.

__init__(client, method, url='', params=None, headers=None, json=None, data=None, cookies=None, files=None)

Initializes the When class with the details of the request to be made.

Parameters:

Name	Type	Description	Default
client	Client	The client instance to use for making the request.	required
method	str	The HTTP method to use for the request.	required
url	str	The URL to send the request to.	''
params	Optional[Dict[str, Any]]	Optional dictionary of query parameters.	None
headers	Optional[Dict[str, Any]]	Optional dictionary of request headers.	None
json	Optional[Any]	Optional JSON body for the request.	None

with_auth(username, password)

Adds basic authentication to the request.

Parameters:

Name	Type	Description	Default
username	str	The username for basic auth.	required
password	str	The password for basic auth.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/basic-auth/user/pass").with_auth("user", "pass").then()...

Returns:

Name	Type	Description
When	When	The instance of the When class.

with_oauth2(token)

Adds OAuth2 authentication to the request.

Parameters:

Name	Type	Description	Default
token	str	The OAuth2 token to use for auth.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> token = "some_token"
>>> given(client).when("GET", "/bearer").with_oauth2(token).then()...

Returns:

Name	Type	Description
When	When	The instance of the When class.

with_api_key(key, value)

Adds API key authentication to the request.

Parameters:

Name	Type	Description	Default
key	str	The key of the API key.	required
value	str	The value of the API key.	required

Returns:

Name	Type	Description
When	When	The instance of the When class.

then(follow_redirects=False, timeout=5.0, force_json_decoding=False)

Transitions from the When stage to the Then stage, where the response is handled.

Parameters:

Name	Type	Description	Default
follow_redirects	bool	httpx parameter to follow redirects or not. Defaults to False.	False
timeout	float	The timeout for the request in seconds. Defaults to 5.0.	5.0
force_json_decoding	bool	If True, forces JSON decoding of the response despite response headers. Defaults to False. The default behavior is to decode JSON only if the response content type is 'application/json'.	False

Note: The actual request is made when this method is called.

Returns:

Name	Type	Description
Then	Then	The instance of the Then class with the response from the request.

then_async(follow_redirects=False, timeout=5.0, force_json_decoding=False) async

Async version of the then method awaiting the response.

Parameters:

Name	Type	Description	Default
follow_redirects	bool	httpx parameter to follow redirects or not. Defaults to False.	False
timeout	float	The timeout for the request in seconds. Defaults to 5.0.	5.0
force_json_decoding	bool	If True, forces JSON decoding of the response despite response headers. Defaults to False. The default behavior is to decode JSON only if the response content type is 'application/json'.	False

Returns:

Name	Type	Description
Then	Then	The instance of the Then class with the response from the request

Then

Represents the Then stage of the request where the response is handled and assertions are made.

__init__(response, client)

Initializes the Then class with the response to handle.

Parameters:

Name	Type	Description	Default
response	UnifiedResponse	The response from the request.	required
client	Client	The client instance used for making the request.	required

get_response()

Retrieves the response object.

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> r = given(client).when("GET", "/get").then().get_response()
>>> r.status_code
>>> 200

Returns:

Name	Type	Description
UnifiedResponse	UnifiedResponse	The response from the request.

validate_data(expected_model)

Validates the response data against the expected Pydantic model.

Parameters:

Name	Type	Description	Default
expected_model	Type[BaseModel]	The Pydantic model to validate the response data against.	required

Raises:

Type	Description
AssertionError	If the response data does not match the expected model.

Examples:

>>> from reqflow import given, Client
>>> from pydantic import BaseModel
>>>
>>> client = Client(base_url="https://httpbin.org")
>>>
>>> class Data(BaseModel):
>>>     url: str
>>>     args: dict
>>>     headers: dict
>>>     origin: str
>>>     method: str
>>>     ...
>>> given(client).when("GET", "/get").then().validate_data(Data)

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

status_code(expected_status_code)

Asserts that the response status code matches the expected status code.

Parameters:

Name	Type	Description	Default
expected_status_code	int	The expected status code of the response.	required

Raises:

Type	Description
AssertionError	If the response status code does not match the expected status code.

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/get").then().status_code(200)

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

status_code_is_between(min_status_code, max_status_code)

Asserts that the response status code is within the specified range.

Parameters:

Name	Type	Description	Default
min_status_code	int	The minimum acceptable status code.	required
max_status_code	int	The maximum acceptable status code.	required

Raises:

Type	Description
AssertionError	If the response status code is not within the specified range.

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/get").then().status_code_is_between(200, 299)

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

assert_body(json_path, expected_value)

Asserts that a specific part of the response body matches the expected value.

Parameters:

Name	Type	Description	Default
json_path	str	The JSONPath expression to locate the part of the response body to assert.	required
expected_value	Any	The expected value to compare against.	required

Raises:

Type	Description
ValueError	If the JSONPath does not match any elements in the JSON response.

Examples:

>>> from reqflow import given, Client
>>> from reqflow.assertions import equal_to
>>>
>>> client = Client(base_url="https://httpbin.org")
>>>
>>> payload = {"foo": "bar"}
>>> given(client).body(payload).when("POST", "/post").then().assert_body("json.foo", equal_to("bar"))

Note

The jsonpath-ng expression is evaluated against the response body as a JSON object.

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

assert_body_text(expected_value)

Asserts that the response body matches the expected value.

Parameters:

Name	Type	Description	Default
expected_value	str	The expected value to compare against.	required

Examples:

>>> from reqflow import given, Client
>>> from reqflow.assertions import equal_to
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/get").then().assert_body_text("Hello, World!")

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

get_content()

Retrieves the content of the response body.

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> r = given(client).when("GET", "/get").then().get_content()

Returns:

Name	Type	Description
Any	Any	The content of the response body.

get_header(header_name)

Retrieves the value of a specific header from the response.

Parameters:

Name	Type	Description	Default
header_name	str	The name of the header to retrieve.	required

Examples >>> from reqflow import given, Client >>> client = Client(base_url="https://httpbin.org") >>> r = given(client).when("GET", "/get").then().get_header("Content-Type") >>> r >>> "application/json"

Returns:

Name	Type	Description
str	Any	The value of the specified header.

get_headers()

Retrieves all headers from the response.

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> r = given(client).when("GET", "/get").then().get_headers()
>>> r
>>> {"Content-Type": "application/json", "Content-Length": "123"}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: A dictionary of all headers in the response.

get_encoding()

Retrieves the encoding of the response.

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> r = given(client).when("GET", "/get").then().get_encoding()
>>> r
>>> "utf-8"

Returns:

Name	Type	Description
str	str	The encoding of the response.

assert_header(header_name, expected_value)

Asserts that a specific header matches the expected value.

Parameters:

Name	Type	Description	Default
header_name	str	The name of the header to assert.	required
expected_value	Any	The expected value of the header.	required

Examples:

>>> from reqflow import given, Client
>>> from reqflow.assertions import equal_to
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/get").then().assert_header("Content-Type", equal_to("application/json"))

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

assert_header_exists(header_name)

Asserts that a specific header exists in the response.

Parameters:

Name	Type	Description	Default
header_name	str	The name of the header to assert.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/get").then().assert_header_exists("Content-Type")

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

assert_response_time(max_time)

Asserts that the response time is less than or equal to the specified maximum time.

Parameters:

Name	Type	Description	Default
max_time	float	The maximum expected response time in seconds.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/get").then().assert_response_time(1.0)

Returns:

Name	Type	Description
Then	Then	The instance of the Then class for fluent chaining.

Raises:

Type	Description
AssertionError	If the response time exceeds the maximum expected time.

assert_cookie(cookie_name, expected_value)

Asserts that a specific cookie matches the expected value.

Parameters:

Name	Type	Description	Default
cookie_name	str	The name of the cookie to assert.	required
expected_value	Any	The expected value of the cookie.	required

Examples:

>>> from reqflow import given, Client
>>> from reqflow.assertions import equal_to
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).query_param('chocolate', 'chip').when("GET", "/cookies/set").then()            >>>                                      .assert_cookie("chocolate", equal_to("chip"))

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

get_cookies()

Retrieves all cookies from the response.

Examples:

>>> client = Client(base_url="https://httpbin.org")
>>> given(client).query_param('chocolate', 'chip').when("GET", "/cookies/set").then().get_cookies()
>>> {'chocolate': 'chip'}

Returns: dict: A dictionary of all cookies in the response.

save_response_to_file(file_path)

Saves the response content to a specified file. Useful for downloading files.

Parameters:

Name	Type	Description	Default
file_path	str	The path where the response content should be saved.	required

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://httpbin.org")
>>> given(client).when("GET", "/image/png").then().save_response_to_file("image.png")

Returns:

Name	Type	Description
Then	Then	The instance of the Then class.

given(client=None, url=None, logging=False)

Initializes the Given stage with a client instance or a URL. If the client is not provided, the URL can be provided directly.

Parameters:

Name	Type	Description	Default
client	Client	The client instance to use for making the request.	None
url	Optional[str]	If the client is not provided, the URL can be provided directly. The client will be initialized with the URL as base_url.	None
logging	bool	If True, logs will be stored in GlobalLogger class.	False

Examples:

>>> from reqflow import given, Client
>>> client = Client(base_url="https://url.com")
>>> given(client).when("GET", "/path").then().status_code(200)
>>> # OR
>>> given(url="https://url.com").when("GET", "/path").then().status_code(200)
>>> # OR
>>> given(url="https://url.com", logging=True).when("GET", "/path").then().status_code(200)

Returns:

Name	Type	Description
Given	class	An instance of the Given class initialized with the provided client.