Client(secret: Optional[str] = None, dry: bool = False, write_delay_ms: int = 1000)[source]¶
The main class for interacting with Beneath. Data-related features (like defining tables and reading/writing data) are implemented directly on Client, while control-plane features (like creating projects) are isolated in the admin member.
secret (str) – A beneath secret to use for authentication. If not set, uses the
BENEATH_SECRETenvironment variable, and if that is not set either, uses the secret authenticated in the CLI (stored in
dry (bool) –
If true, the client will not perform any mutations or writes, but generally perform reads as usual. It’s useful for testing.
The exact implication differs for different operations: Some mutations will be mocked, such as creating a table, others will fail with an exception. Write operations log records to the logger instead of transmitting to the server. Reads generally work, but throw an exception when called on mocked resources.
write_delay_ms (int) – The maximum amount of time to buffer written records before sending a batch write request over the network. Defaults to 1 second (1000 ms). Writing records in batches reduces the number of requests, which leads to lower cost (Beneath charges at least 1kb per request).
find_table(table_path: str) → beneath.table.Table[source]¶
Finds an existing table and returns an object that you can use to read and write from/to the table.
path (str) – The path to the table in the format of “USERNAME/PROJECT/TABLE”
create_table(table_path: str, schema: str, description: Optional[str] = None, meta: Optional[bool] = None, use_index: Optional[bool] = None, use_warehouse: Optional[bool] = None, retention: Optional[datetime.timedelta] = None, log_retention: Optional[datetime.timedelta] = None, index_retention: Optional[datetime.timedelta] = None, warehouse_retention: Optional[datetime.timedelta] = None, schema_kind: str = 'GraphQL', indexes: Optional[str] = None, update_if_exists: Optional[bool] = None) → beneath.table.Table[source]¶
Creates (or optionally updates if
update_if_exists=True) a table and returns it.
table_path (str) – The (desired) path to the table in the format of “USERNAME/PROJECT/TABLE”. The project must already exist. If the table doesn’t exist yet, it creates it.
schema (str) – The GraphQL schema for the table. To learn about the schema definition language, see https://about.beneath.dev/docs/reading-writing-data/schema-definition/.
description (str) – The description shown for the table in the web console. If not set, tries to infer a description from the schema.
retention (timedelta) – The amount of time to retain records written to the table. If not set, records will be stored forever.
schema_kind (str) – The parser to use for
schema. Currently must be “GraphQL” (default).
update_if_exists (bool) – If true and the table already exists, the provided info will be used to update the table (only supports non-breaking schema changes) before returning it.
query_warehouse(query: str, analyze: bool = False, max_bytes_scanned: int = 10000000000, timeout_ms: int = 60000)[source]¶
Starts a warehouse (OLAP) SQL query, and returns a job for tracking its progress
query (str) – The analytical SQL query to run. To learn about the query language, see https://about.beneath.dev/docs/reading-writing-data/warehouse-queries/.
analyze (bool) – If true, analyzes the query and returns info about referenced tables and expected bytes scanned, but doesn’t actually run the query.
max_bytes_scanned (int) – Sets a limit on the number of bytes the query can scan. If exceeded, the job will fail with an error.
Opens the client for writes. Can be called multiple times, but make sure to call
Closes the client for writes, ensuring buffered writes are flushed. If
startwas called multiple times, only the last corresponding call to
stoptriggers a flush.
write(instance: beneath.instance.TableInstance, records: Union[collections.abc.Mapping, Iterable[collections.abc.Mapping]])[source]¶
Writes one or more records to
instance. By default, writes are buffered for up to
write_delay_msmilliseconds before being transmitted to the server. See the Client constructor for details.
To enabled writes, make sure to call
starton the client (and
instance (TableInstance) – The instance to write to. You can also call
instance.writeas a convenience wrapper.
records – The records to write. Can be a single record (dict) or a list of records (iterable of dict).
write_full(table_path: str, records: Union[Iterable[dict], pandas.core.frame.DataFrame], key: Optional[Union[str, List[str]]] = None, description: Optional[str] = None, recreate_on_schema_change=False)[source]¶
Infers a schema, creates a table, and writes a full dataset to Beneath. Each call will create a new primary version for the table, and delete the old primary version if/when the write completes succesfully.
table_path (str) – The (desired) path to the table in the format of “USERNAME/PROJECT/TABLE”. The project must already exist.
records (list(dict) | pandas.DataFrame) – The full dataset to write, either as a list of records or as a Pandas DataFrame. This function uses
beneath.infer_avroto infer a schema for the table based on the records.
key (str | list(str)) – The fields to use as the table’s key. If not set, will default to the dataframe index if
recordsis a Pandas DataFrame, or add a column of incrementing numbers if
recordsis a list.
description (str) – A description for the table.
recreate_on_schema_change (bool) – If true, and there’s an existing table at
table_pathwith a schema that is incompatible with the inferred schema for
records, it will delete the existing table and create a new one instead of throwing an error. Defaults to false.
checkpointer(project_path: str, key_prefix: Optional[str] = None, metatable_name='checkpoints', metatable_create: bool = True, metatable_description='Stores checkpointed state for consumers, pipelines, and more') → beneath.checkpointer.Checkpointer[source]¶
Returns a checkpointer for the given project. Checkpointers store (small) key-value records useful for maintaining consumer and pipeline state. State is stored in a meta-table called “checkpoints” in the given project.
project_path (str) – Path to the project in which to store the checkpointer’s state
key_prefix (str) – If set, any
setcall on the checkpointer will prepend the prefix to the key.
metatable_name (str) – Name of the meta table in which to save checkpointed data
metatable_create (bool) – If true, the checkpointer will create the checkpoints meta-table if it does not already exists. If false, the checkpointer will throw an exception if the meta-table does not already exist. Defaults to True.
metatable_description (str) – An optional description to apply to the checkpoints meta-table. Defaults to a sensible description of checkpointing.
consumer(table_path: str, version: Optional[int] = None, batch_size: int = 1000, subscription_path: Optional[str] = None, checkpointer: Optional[beneath.checkpointer.Checkpointer] = None, metatable_create: bool = True)[source]¶
Creates a consumer for the given table. Consumers make it easy to replay the history of a table and/or subscribe to new changes.
table_path (str) – Path to the table to subscribe to. The consumer will subscribe to the table’s primary version.
version (int) – The instance version to use for table. If not set, uses the primary instance.
batch_size (int) – Sets the max number of records to load in each network request. Defaults to 1000.
subscription_path (str) – Format “ORGANIZATION/PROJECT/NAME”. If set, the consumer will use a checkpointer to save cursors. That means processing will not restart from scratch if the process ends or crashes (as long as you use the same subscription name). To reset a subscription, call
reseton the consumer.
checkpointer (Checkpointer) – Only applies if
subscription_pathis set. Provides a specific checkpointer to use for consumer state. If not set, will create one in the subscription’s project.
metatable_create (bool) – Only applies if
subscription_pathis set and
checkpointeris not set. Passed through to