stigg.generated.AsyncBaseClient

class documentation

class AsyncBaseClient:

Known subclasses: stigg.generated.AsyncClient

Constructor: AsyncBaseClient(url, headers, http_client, ws_url, ...)

View In Hierarchy

Undocumented

Async Method __aenter__ Undocumented
Async Method __aexit__ Undocumented
Method __init__ Undocumented
Async Method execute Undocumented
Async Method execute_ws Undocumented
Method get_data Undocumented
Instance Variable headers Undocumented
Instance Variable http_client Undocumented
Instance Variable url Undocumented
Instance Variable ws_connection_init_payload Undocumented
Instance Variable ws_headers Undocumented
Instance Variable ws_origin Undocumented
Instance Variable ws_url Undocumented
Method _convert_dict_to_json_serializable Undocumented
Method _convert_value Undocumented
Async Method _execute_json Undocumented
Async Method _execute_multipart Undocumented
Method _get_files_from_variables Undocumented
Async Method _handle_ws_message Undocumented
Method _process_variables Undocumented
Async Method _send_connection_init Undocumented
Async Method _send_subscribe Undocumented
async def __aenter__(self: Self) -> Self:

Undocumented

async def __aexit__(self, exc_type: object, exc_val: object, exc_tb: object):

Undocumented

def __init__(self, url: str = '', headers: dict[str, str] | None = None, http_client: httpx.AsyncClient | None = None, ws_url: str = '', ws_headers: dict[str, Any] | None = None, ws_origin: str | None = None, ws_connection_init_payload: dict[str, Any] | None = None):

Undocumented

async def execute(self, query: str, operation_name: str | None = None, variables: dict[str, Any] | None = None, **kwargs: Any) -> httpx.Response:

Undocumented

async def execute_ws(self, query: str, operation_name: str | None = None, variables: dict[str, Any] | None = None, **kwargs: Any) -> AsyncIterator[dict[str, Any]]:

Undocumented

def get_data(self, response: httpx.Response) -> dict[str, Any]:

Undocumented

headers =

Undocumented

http_client =

Undocumented

url =

Undocumented

ws_connection_init_payload =

Undocumented

ws_headers =

Undocumented

ws_origin =

Undocumented

ws_url =

Undocumented

def _convert_dict_to_json_serializable(self, dict_: dict[str, Any]) -> dict[str, Any]:

Undocumented

def _convert_value(self, value: Any) -> Any:

Undocumented

async def _execute_json(self, query: str, operation_name: str | None, variables: dict[str, Any], **kwargs: Any) -> httpx.Response:

Undocumented

async def _execute_multipart(self, query: str, operation_name: str | None, variables: dict[str, Any], files: dict[str, tuple[str, IO[bytes], str]], files_map: dict[str, list[str]], **kwargs: Any) -> httpx.Response:

Undocumented

def _get_files_from_variables(self, variables: dict[str, Any]) -> tuple[dict[str, Any], dict[str, tuple[str, IO[bytes], str]], dict[str, list[str]]]:

Undocumented

async def _handle_ws_message(self, message: Data, websocket: WebSocketClientProtocol, expected_type: GraphQLTransportWSMessageType | None = None) -> dict[str, Any] | None:

Undocumented

def _process_variables(self, variables: dict[str, Any] | None) -> tuple[dict[str, Any], dict[str, tuple[str, IO[bytes], str]], dict[str, list[str]]]:

Undocumented

async def _send_connection_init(self, websocket: WebSocketClientProtocol):

Undocumented

async def _send_subscribe(self, websocket: WebSocketClientProtocol, operation_id: str, query: str, operation_name: str | None = None, variables: dict[str, Any] | None = None):

Undocumented