pypergraph.core.cross_platform package

Subpackages

• pypergraph.core.cross_platform.di package
  • Submodules
  • pypergraph.core.cross_platform.di.json_storage module
    • JsonStorage
      • JsonStorage.get_item()
      • JsonStorage.remove_item()
      • JsonStorage.set_item()
  • pypergraph.core.cross_platform.di.rest_client module
    • HttpxClient
      • HttpxClient.close()
      • HttpxClient.request()
    • RESTClient
      • RESTClient.close()
      • RESTClient.request()
  • Module contents

Submodules

pypergraph.core.cross_platform.rest_api_client module

class pypergraph.core.cross_platform.rest_api_client.RestAPIClient(base_url: str, client: RESTClient | None = None, timeout: int = 30)

Bases: object

property base_url: str

Returns the base URL.

async close()

Closes the HTTPClient session if it was not provided by the user.

config(client: RESTClient)

async delete(endpoint: str, headers: Dict[str, str] | None = None, params: Dict[str, Any] | None = None)

async get(endpoint: str, headers: Dict[str, str] | None = None, params: Dict[str, Any] | None = None)

handle_api_response(response: Response, method: str, endpoint: str)

Handles API responses, checking for errors and returning JSON or text.

async post(endpoint: str, headers: Dict[str, str] | None = None, payload: Dict[str, Any] | None = None)

async put(endpoint: str, headers: Dict[str, str] | None = None, payload: Dict[str, Any] | None = None)

async request(method: str, endpoint: str, headers: Dict[str, str] | None = None, params: Dict[str, Any] | None = None, payload: Dict[str, Any] | None = None)

Makes an HTTP request.

pypergraph.core.cross_platform.state_storage_db module

class pypergraph.core.cross_platform.state_storage_db.StateStorageDb(storage_client=None, file_path: str | None = None)

Bases: object

async delete(key: str = 'vault')

async get(key: str = 'vault')

async set(key: str | None, value: any)

set_client(client)

set_prefix(prefix: str)

Module contents