Skip to content

Requests

Create request

Models

API

API zapy.requests.models.RequestMetadata
Class

RequestMetadata( cell_type = 'zapy.ZapyRequest', v = <sdk_version>, tags=[] )

Attributes
cell_type : str = 'zapy.ZapyRequest'
Cell type, use zapy.ZapyRequest for requests
v : str = <sdk_version>
Version of the SDK
tags : list[str] = []
Optional tags
API zapy.requests.models.ZapyRequest
Class

ZapyRequest( endpoint, method, **kwargs )

Attributes

endpoint : str

method : str

metadata : RequestMetadata = None

variables : list[KeyValueItem] = []

params : list[KeyValueItem] = []

headers : list[KeyValueItem] = []

script : list[str] = []

body_type : str = 'None'

body : list[str] | list[KeyValueItem] | None = None

API zapy.requests.models.KeyValueItem
Class

KeyValueItem( key, value, active=True )

Attributes

key : str

value : str

active : bool = True

Example:

from zapy.requests.models import ZapyRequest, KeyValueItem, RequestMetadata

zapy_request = ZapyRequest(
    endpoint="http://test/",
    method="GET",
    metadata=RequestMetadata(
        cell_type='zapy.ZapyRequest',
        v='0.0.1',
        tags=['tag_1'],
    ),
    variables=[
        KeyValueItem(key='var1', value='Hello', active=True),
    ],
    params=[
        KeyValueItem(key='param1', value='{{var1 + "World"}}', active=True),
    ],
    headers=[
        KeyValueItem(key='Content-Type', value='{{ ctx.auto() }}', active=True),
    ],
    script=[
        'from decimal import Decimal',
    ],
    body_type='application/json',
    body=[
        '{',
        '   hello": "world"'
        '}',
    ],
)

Import request

API

API zapy.requests.from_file
Function

from_file( path )

Arguments
path : Path | str
Returns
ZapyRequest
from zapy import requests
from zapy.requests.models import ZapyRequest

req: ZapyRequest = requests.from_file('my_zapy.zapy')

Send request

ZapyRequest.send

Import or create requests and send them. ZapyRequest.send may throw throw error test.

API ZapyRequest.send Async Method
[async] send( store=ZapyAuto, logger=print, raise_assert=True )
Arguments
store : Store | ZapyAuto=ZapyAuto
  • If ZapyAuto, then it will use the default store
  • Use a zapy.store.Store to define a custom store
logger : print = print
print or print-like function
raise_assert : AssertTestResultMixin | bool = True
  • If True, then it will raise an assertion exception if any test fails.
  • If False, then it will not raise an exception on test even if there are failures.
  • If it's called inside a unittest.TestCase class you can extend the class with AssertTestResultMixin and pass self to raise_assert in order to improve the assertion error message.
Returns
httpx.Response
from zapy import requests

req = requests.from_file('my_zapy.zapy')
httpx_response = await req.send()

requests.send_request

will not throw throw error test

API zapy.requests.send_request
Async Function

[async] send_request( store=ZapyAuto, logger=print )

Arguments
store : Store | ZapyAuto=ZapyAuto
  • If ZapyAuto, then it will use the default store
  • Use a zapy.store.Store to define a custom store
logger : print = print
print or print-like function
Returns
zapy.requests.RequesterResponse
API zapy.requests.RequesterResponse
Class

[async] send_request( store=ZapyAuto, logger=print )

Attributes

response : httpx.Response

test_result : zapy.test.models.TestResultDict = None

from zapy import requests
from zapy.requests import send_request
from zapy.test import assert_test_result_dict

req = requests.from_file('my_zapy.zapy')
response_wrapper = await send_request(req)
assert_test_result_dict(response_wrapper.test_result)

Execute chain

Load a chain to execute it and use the variables created.

API zapy.utils.module.load_ipynb
Async Function

[async] load_ipynb( path, variables=None )

Arguments
load_ipynb : Path | str
The path to the .ipynb file
variables : dict | None = None
print or print-like function
Returns
types.ModuleType
Module that contains generated variables in .ipynb file
from zapy.utils import module
rel_path = Path('tests/assets')
# invoke jupyter notebook
chain = await module.load_ipynb(rel_path / 'chain_import.ipynb', variables={
    'rel_path': rel_path
})
# access a variable
print(chain.out_response)

Reusing connection

Not available for now.