1. [#528](https://github.com/influxdata/influxdb-client-python/pull/528): Add `BucketSchemasService` to manage explicit bucket schemas to enforce column names, tags, fields, and data types for your data

:warning: The following examples are related to [InfluxDB Cloud](https://docs.influxdata.com/influxdb/cloud/) and not available on a local InfluxDB OSS instance.

- [invokable_scripts.py](invokable_scripts.py) - How to use Invokable scripts Cloud API to create custom endpoints that query data

- [bucket_schemas.py](bucket_schemas.py) - How to manage explicit bucket schemas to enforce column names, tags, fields, and data types for your data

import datetime

from influxdb_client import InfluxDBClient, BucketSchemasService, PostBucketRequest, SchemaType, \

MeasurementSchemaCreateRequest, MeasurementSchemaColumn, ColumnSemanticType, ColumnDataType, \

MeasurementSchemaUpdateRequest

influx_cloud_url = 'https://us-west-2-1.aws.cloud2.influxdata.com'

influx_cloud_token = '...'

org_name = '...'

with InfluxDBClient(url=influx_cloud_url, token=influx_cloud_token, org=org_name, debug=False) as client:

uniqueId = str(datetime.datetime.now())

org_id = client.organizations_api().find_organizations(org=org_name)[0].id

bucket_schemas_api = BucketSchemasService(api_client=client.api_client)

"""

print("------- Create Bucket -------\n")

created_bucket = client \

.buckets_api() \

.create_bucket(bucket=PostBucketRequest(name=f"my_schema_bucket_{uniqueId}",

org_id=org_id,

retention_rules=[],

schema_type=SchemaType.EXPLICIT))

print(created_bucket)

"""

print("------- Create Schema -------\n")

columns = [

MeasurementSchemaColumn(name="time",

type=ColumnSemanticType.TIMESTAMP),

MeasurementSchemaColumn(name="service",

type=ColumnSemanticType.TAG),

MeasurementSchemaColumn(name="host",

type=ColumnSemanticType.TAG),

MeasurementSchemaColumn(name="usage_user",

type=ColumnSemanticType.FIELD,

data_type=ColumnDataType.FLOAT),

MeasurementSchemaColumn(name="usage_system",

type=ColumnSemanticType.FIELD,

data_type=ColumnDataType.FLOAT)

]

create_request = MeasurementSchemaCreateRequest(name="usage_cpu", columns=columns)

created_schema = bucket_schemas_api.create_measurement_schema(bucket_id=created_bucket.id,

org_id=org_id,

measurement_schema_create_request=create_request)

print(created_bucket)

"""

print("\n------- Lists the Schemas -------\n")

measurement_schemas = bucket_schemas_api.get_measurement_schemas(bucket_id=created_bucket.id).measurement_schemas

print("\n".join([f"---\n ID: {ms.id}\n Name: {ms.name}\n Columns: {ms.columns}" for ms in measurement_schemas]))

print("---")

"""

print("------- Update a bucket schema -------\n")

columns.append(MeasurementSchemaColumn(name="usage_total",

type=ColumnSemanticType.FIELD,

data_type=ColumnDataType.FLOAT))

update_request = MeasurementSchemaUpdateRequest(columns=columns)

updated_schema = bucket_schemas_api.update_measurement_schema(bucket_id=created_bucket.id,

measurement_id=created_schema.id,

measurement_schema_update_request=update_request)

print(updated_schema)

"""

print("------- Delete Bucket -------\n")

client.buckets_api().delete_bucket(created_bucket)

print(f" successfully deleted bucket: {created_bucket.name}")

from influxdb_client.service.bucket_schemas_service import BucketSchemasService

from influxdb_client.domain.column_data_type import ColumnDataType

from influxdb_client.domain.column_semantic_type import ColumnSemanticType

from influxdb_client.domain.measurement_schema import MeasurementSchema

from influxdb_client.domain.measurement_schema_column import MeasurementSchemaColumn

from influxdb_client.domain.measurement_schema_create_request import MeasurementSchemaCreateRequest

from influxdb_client.domain.measurement_schema_list import MeasurementSchemaList

from influxdb_client.domain.measurement_schema_update_request import MeasurementSchemaUpdateRequest

from influxdb_client.service.bucket_schemas_service import BucketSchemasService

return ping_service.response_header(response)

def build(self) -> str:

"""

ping_service = PingService(self.api_client)

return ping_service.build_type()

return ping_service.response_header(response)

async def build(self) -> str:

"""

ping_service = PingService(self.api_client)

return await ping_service.build_type_async()

class CloudOnlyWarning(UserWarning):

"""User warning about availability only on the InfluxDB Cloud."""

@staticmethod

def print_warning(api_name: str, doc_url: str):

"""Print warning about availability only on the InfluxDB Cloud."""

message = f"""The '{api_name}' is available only on the InfluxDB Cloud.

warnings.warn(message, CloudOnlyWarning)

from influxdb_client.service.bucket_schemas_service import BucketSchemasService

from influxdb_client.domain.column_data_type import ColumnDataType

from influxdb_client.domain.column_semantic_type import ColumnSemanticType

from influxdb_client.domain.measurement_schema import MeasurementSchema

from influxdb_client.domain.measurement_schema_column import MeasurementSchemaColumn

from influxdb_client.domain.measurement_schema_create_request import MeasurementSchemaCreateRequest

from influxdb_client.domain.measurement_schema_list import MeasurementSchemaList

from influxdb_client.domain.measurement_schema_update_request import MeasurementSchemaUpdateRequest

import pprint

import re # noqa: F401

class ColumnDataType(object):

"""NOTE: This class is auto generated by OpenAPI Generator.

"""

INTEGER = "integer"

FLOAT = "float"

BOOLEAN = "boolean"

STRING = "string"

UNSIGNED = "unsigned"

"""

openapi_types = {

}

attribute_map = {

}

def __init__(self): # noqa: E501,D401,D403

"""ColumnDataType - a model defined in OpenAPI.""" # noqa: E501 self.discriminator = None

def to_dict(self):

"""Return the model properties as a dict."""

result = {}

for attr, _ in self.openapi_types.items():

value = getattr(self, attr)

if isinstance(value, list):

result[attr] = list(map(

lambda x: x.to_dict() if hasattr(x, "to_dict") else x,

value

))

elif hasattr(value, "to_dict"):

result[attr] = value.to_dict()

elif isinstance(value, dict):

result[attr] = dict(map(

lambda item: (item[0], item[1].to_dict())

if hasattr(item[1], "to_dict") else item,

value.items()

))

else:

result[attr] = value

return result

def to_str(self):

"""Return the string representation of the model."""

return pprint.pformat(self.to_dict())

def __repr__(self):

"""For `print` and `pprint`."""

return self.to_str()

def __eq__(self, other):

"""Return true if both objects are equal."""

if not isinstance(other, ColumnDataType):

return False

return self.__dict__ == other.__dict__

def __ne__(self, other):

"""Return true if both objects are not equal."""

return not self == other