Source code for influxdb_client.client.bucket_api

A bucket is a named location where time series data is stored.

All buckets have a retention policy, a duration of time that each data point persists.
A bucket belongs to an organization.

from influxdb_client import BucketsService, Bucket, PostBucketRequest

[docs]class BucketsApi(object): """Implementation for '/api/v2/buckets' endpoint.""" def __init__(self, influxdb_client): """Initialize defaults.""" self._influxdb_client = influxdb_client self._buckets_service = BucketsService(influxdb_client.api_client)
[docs] def create_bucket(self, bucket=None, bucket_name=None, org_id=None, retention_rules=None, description=None) -> Bucket: """Create a bucket. :param Bucket bucket: bucket to create (required) :param bucket_name: bucket name :param description: bucket description :param org_id: org_id :param bucket_name: bucket name :param retention_rules: retention rules array or single BucketRetentionRules :return: Bucket If the method is called asynchronously, returns the request thread. """ if retention_rules is None: retention_rules = [] rules = [] if isinstance(retention_rules, list): rules.extend(retention_rules) else: rules.append(retention_rules) if bucket is None: bucket = PostBucketRequest(name=bucket_name, retention_rules=rules, description=description) if org_id is None: org_id = bucket.org_id = org_id return self._buckets_service.post_buckets(post_bucket_request=bucket)
[docs] def delete_bucket(self, bucket): """Delete a bucket. :param bucket: bucket id or Bucket :return: Bucket If the method is called asynchronously, returns the request thread. """ if isinstance(bucket, Bucket): bucket_id = else: bucket_id = bucket return self._buckets_service.delete_buckets_id(bucket_id=bucket_id)
[docs] def find_bucket_by_id(self, id): """Find bucket by ID. :param id: :return: """ return self._buckets_service.get_buckets_id(id)
[docs] def find_bucket_by_name(self, bucket_name): """Find bucket by name. :param bucket_name: bucket name :return: Bucket """ buckets = self._buckets_service.get_buckets(name=bucket_name) if len(buckets.buckets) > 0: return buckets.buckets[0] else: return None
[docs] def find_buckets(self): """Get all buckets.""" return self._buckets_service.get_buckets()