Client

pingthings.timeseries.client.Client ¶

Client(loop: ClientEventLoop, client: AsyncClient)

Class that manages reusable state and client connections.

This class provides synchronous methods to connect to an asynchronous client, retrieve client and event loop information, and perform various client-related operations such as querying and streaming.

Info

For most users, this client connection will be the only necessary connection type to work with. Queries are accelerated using the asynchronous methods under the hood. If for some reason you need to use the asynchronous client, refer to pingthings.timeseries.async_client.

Create a synchronous client class that manages state for asynchronous client connections.

Parameters:

Name	Type	Description	Default
`loop` ¶	`ClientEventLoop`	The event loop used by the client.	required
`client` ¶	`AsyncClient`	The asynchronous client instance.	required

Methods:

Name	Description
`connect`	Establish a synchronous connection to the asynchronous client.
`create`	Create a new stream.
`get_async_client`	Retrieve the asynchronous client instance.
`get_event_loop`	Retrieve the event loop instance.
`info`	Retrieve information about the server and proxy server the client is connected to.
`list_collections`	Returns a list of collection paths using the `prefix` argument for
`sql_query`	Performs a SQL query on the database metadata and returns a list of
`stream_from_uuid`	Retrieve a stream based on its UUID.
`streams_in_collection`	Search for streams matching given parameters
`streamset_from_uuids`	Return a `StreamSet` from an iterable of UUIDs.

Attributes:

Name	Type	Description
`concurrency_limit`		The concurrency limit for background asynchronous operations.

Attributes¶

concurrency_limit `property` `writable` ¶

concurrency_limit

The concurrency limit for background asynchronous operations.

Setting the value overrides the environment variable

If you choose to set a custom concurrency_limit, this will bypass the environment variable PINGTHINGS_CONCURRENCY_LIMIT value.

Functions¶

connect `staticmethod` ¶

connect(
    profile: Optional[str] = None,
    endpoint: Optional[str] = None,
    apikey: Optional[str] = None,
    concurrency_limit: Optional[int] = None,
) -> Client

Establish a synchronous connection to the asynchronous client.

Parameters:

Name	Type	Description	Default
`profile` ¶	`Optional[str]`	The name of a profile containing the required connection information as found in the user's predictive grid credentials file `${HOME}/.predictivegrid/credentials.yaml`.	`None`
`endpoint` ¶	`Optional[str]`	The address and port of the cluster to connect to, e.g. `192.168.1.1:4411`, if not set, will look for the environment variable `$BTRDB_ENDPOINTS`	`None`
`apikey` ¶	`Optional[str]`	The API key used to authenticate requests, if not set, the key is looked up from the environment variable `$BTRDB_API_KEY`.	`None`
`concurrency_limit` ¶	`Optional[int]`	The maximum number of concurrent database requests to have in flight at any one time, if not set, will be inferred from environment variable `$PINGTHINGS_CONCURRENCY_LIMIT`.	`None`

Returns:

Type	Description
`Client`	An instance of the Client class.

Examples:

Connecting to the timeseries platform as a commercial customer in the PingThings provided JupyterHub/Lab environment. This behavior also works if you have the environment variables set, refer to the above docstring for more information.

import pingthings as pt
conn = pt.timeseries.connect()

Connecting to the timeseries platform when you know your api key and FQDN endpoint.

import pingthings as pt
my_key = "ABC123"
my_endpoint = "example.com:4411"

conn = pt.timeseries.connect(apikey=my_key, endpoint=my_endpoint)

Connecting to the platform when you have a populated ${HOME}/.predictivegrid/credentials.yaml file with profiles.

import pingthings as pt
conn = pt.timeseries.connect(profile='my_server')

create ¶

create(
    uuid: UUID,
    collection: str,
    tags: Optional[dict[str, str]] = None,
    annotations: Optional[dict[str, str]] = None,
) -> Stream

Create a new stream.

Parameters:

Name	Type	Description	Default
`uuid` ¶	`UUID`	The UUID for the new stream.	required
`collection` ¶	`str`	The collection to which the stream belongs.	required
`tags` ¶	`Optional[dict[str, str]]`	Tags associated with the stream.	`None`
`annotations` ¶	`Optional[dict[str, str]]`	Annotations for the stream.	`None`

Returns:

Type	Description
`Stream`	The newly created stream.

get_async_client ¶

get_async_client() -> AsyncClient

Retrieve the asynchronous client instance.

Returns:

Type	Description
`AsyncClient`	The asynchronous client instance.

get_event_loop ¶

get_event_loop() -> ClientEventLoop

Retrieve the event loop instance.

Returns:

Type	Description
`ClientEventLoop`	The event loop instance used by the client.

info ¶

info() -> dict[str, Any]

Retrieve information about the server and proxy server the client is connected to.

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing server and proxy server information.

list_collections ¶

list_collections(prefix: Optional[str] = None) -> list[str]

Returns a list of collection paths using the prefix argument for filtering.

Parameters:

Name	Type	Description	Default
`prefix` ¶		Filter collections that start with the string provided, if none passed, will list all collections.	`None`

Returns:

Type	Description
`list[str]`	All collections that match the provided prefix.

Examples:

Assuming we have the following collections in the platform: foo, bar, foo/baz, bar/baz

>>> conn = pt.connect()
>>> conn.list_collections().sort()
["bar", "bar/baz", "foo", "foo/bar"]

>>> conn.list_collections(prefix="foo")
["foo", "foo/bar"]

sql_query ¶

sql_query(
    query: str, params: Optional[list[str]] = None
) -> list[Any]

Performs a SQL query on the database metadata and returns a list of dictionaries from the resulting cursor.

Parameters:

Name	Type	Description	Default
`query` ¶	`str`	A SQL statement to be executed on the BTrDB metadata. Available columns in the `stream` table are noted below. To sanitize inputs use a `$1` style parameter such as `select * from streams where name = $1 or name = $2`.	required
`params` ¶	`Optional[list[str]]`	A list of parameter values to be sanitized and interpolated into the SQL statement. Using parameters forces value/type checking and is considered a best practice at the very least.	`None`

Returns:

Type	Description
`list[Any]`	The result of the SQL query.

Available columns in the stream table

column_name	data_type
uuid	uuid
collection	character varying
name	character varying
unit	character varying
ingress	character varying
property_version	bigint
annotations	hstore
distiller	character varying
created_at	timestamp with time zone
updated_at	timestamp with time zone
geo	postgis geometry
last_written	timestamp with time zone
previous_last_written	timestamp with time zone
estimated_count_delta	bigint
long_term_autoregressive_average	double precision
count_last_updated	timestamp with time zone
previous_count_last_updated	timestamp with time zone
watched	boolean

stream_from_uuid ¶

stream_from_uuid(uuid: UUID | str) -> Stream

Retrieve a stream based on its UUID.

Parameters:

Name	Type	Description	Default
`uuid` ¶	`UUID \| str`	The UUID of the stream.	required

Returns:

Type	Description
`Stream`	The stream associated with the provided UUID.

Raises:

Type	Description
`TypeError`	If the provided `uuid` is not a valid UUID

streams_in_collection ¶

streams_in_collection(
    collection: str = "",
    is_collection_prefix: bool = True,
    tags: Optional[dict[str, str]] = None,
    annotations: Optional[dict[str, Any]] = None,
) -> StreamSet

Search for streams matching given parameters

Parameters:

Name	Type	Description	Default
`collection` ¶	`str`	collections to use when searching for streams, case sensitive.	`''`
`is_collection_prefix` ¶	`bool`	Whether the collection is a prefix of the whole collection name.	`True`
`tags` ¶	`Optional[dict[str, str]]`	The tags to identify the stream.	`None`
`annotations` ¶	`Optional[dict[str, Any]]`	The annotations to identify the stream.	`None`

Returns:

Type	Description
`StreamSet`	The grouping of streams matching given parameters.

streamset_from_uuids ¶

streamset_from_uuids(
    uuids: list[UUID | str], fetch_metadata: bool = True
) -> StreamSet

Return a StreamSet from an iterable of UUIDs.

Parameters:

Name	Type	Description	Default
`uuids` ¶	`list[UUID \| str]`	List of stream identifiers	required
`fetch_metadata` ¶	`bool`	Whether to fetch metadata for the streams in the set. Default is True.	`True`

Advanced user feature

Be cautious about using fetch_metadata=False. Many stream metadata values like collection, name, unit, tags, annotations will not be available, meaning filtering and other operations that require metadata will not work.

Returns:

Type	Description
`StreamSet`	The `StreamSet` associated with the provided iterable of UUIDs.

pingthings.timeseries.client.ClientEventLoop ¶

ClientEventLoop()

The object responsible for task running.

This leverages the asyncio event loop.

Are you sure you need to manually use this?

This object is automatically created and leveraged whenever you use the standard connect or async_connect

Methods:

Name	Description
`run_coroutine_threadsafe`	A wrapper to run coroutines threadsafe.

Functions¶

run_coroutine_threadsafe ¶

run_coroutine_threadsafe(
    coro: Coroutine[Any, Any, CO_RESULT_TYPE]
) -> Future[CO_RESULT_TYPE]

A wrapper to run coroutines threadsafe.

Client

pingthings.timeseries.client.Client ¶

`loop` ¶

`client` ¶

Attributes¶

concurrency_limit `property` `writable` ¶

Functions¶

connect `staticmethod` ¶

`profile` ¶

`endpoint` ¶

`apikey` ¶

`concurrency_limit` ¶

create ¶

`uuid` ¶

`collection` ¶

`tags` ¶

`annotations` ¶

get_async_client ¶

get_event_loop ¶

info ¶

list_collections ¶

`prefix` ¶

sql_query ¶

`query` ¶

`params` ¶

stream_from_uuid ¶

`uuid` ¶

streams_in_collection ¶

`collection` ¶

`is_collection_prefix` ¶

`tags` ¶

`annotations` ¶

streamset_from_uuids ¶

`uuids` ¶

`fetch_metadata` ¶

pingthings.timeseries.client.ClientEventLoop ¶

Functions¶

run_coroutine_threadsafe ¶

Client

pingthings.timeseries.client.Client ¶

loop ¶

client ¶

Attributes¶

concurrency_limit property writable ¶

Functions¶

connect staticmethod ¶

profile ¶

endpoint ¶

apikey ¶

concurrency_limit ¶

create ¶

uuid ¶

collection ¶

tags ¶

annotations ¶

get_async_client ¶

get_event_loop ¶

info ¶

list_collections ¶

prefix ¶

sql_query ¶

query ¶

params ¶

stream_from_uuid ¶

uuid ¶

streams_in_collection ¶

collection ¶

is_collection_prefix ¶

tags ¶

annotations ¶

streamset_from_uuids ¶

uuids ¶

fetch_metadata ¶

pingthings.timeseries.client.ClientEventLoop ¶

Functions¶

run_coroutine_threadsafe ¶

`loop` ¶

`client` ¶

concurrency_limit `property` `writable` ¶

connect `staticmethod` ¶

`profile` ¶

`endpoint` ¶

`apikey` ¶

`concurrency_limit` ¶

`uuid` ¶

`collection` ¶

`tags` ¶

`annotations` ¶

`prefix` ¶

`query` ¶

`params` ¶

`uuid` ¶

`collection` ¶

`is_collection_prefix` ¶

`tags` ¶

`annotations` ¶

`uuids` ¶

`fetch_metadata` ¶