docs: add an example how to check connection credentials (#269)

Author: bednar

.circleci/config.yml

@@ -138,6 +138,7 @@ jobs:
             python ./asynchronous_management.py
             python ./asynchronous_batching.py
             python ./asynchronous_retry.py
+            python ./connection_check.py
   check-sphinx:
     docker:
       - image: *default-python

CHANGELOG.md

@@ -15,6 +15,7 @@
 
 ### Documentation
 1. [#397](https://github.com/influxdata/influxdb-client-python/pull/397): Add an example: How to use RxPY to prepare batches by maximum bytes count
+1. [#269](https://github.com/influxdata/influxdb-client-python/pull/269): Add new example: How to check connection credentials
 
 ## 1.31.0 [2022-07-29]
 

examples/README.md

@@ -34,6 +34,7 @@
 - [influxdb_18_example.py](influxdb_18_example.py) - How to connect to InfluxDB 1.8
 - [nanosecond_precision.py](nanosecond_precision.py) - How to use nanoseconds precision
 - [invokable_scripts.py](invokable_scripts.py) - How to use Invokable scripts Cloud API to create custom endpoints that query data
+- [connection_check.py](connection_check.py) - How to check connection configuration
 
 ## Asynchronous
 - [asynchronous.py](asynchronous.py) - How to use Asyncio with InfluxDB client

examples/connection_check.py

@@ -0,0 +1,61 @@
+"""
+How to check that connection credentials are suitable for queries and writes from/into specified bucket.
+"""
+
+from influxdb_client import InfluxDBClient
+from influxdb_client.client.write_api import SYNCHRONOUS
+from influxdb_client.rest import ApiException
+
+"""
+Define credentials
+"""
+url = "http://localhost:8086"
+token = "my-token"
+org = "my-org"
+bucket = "my-bucket"
+
+
+def check_connection():
+    """Check that the InfluxDB is running."""
+    print("> Checking connection ...", end=" ")
+    client.api_client.call_api('/ping', 'GET')
+    print("ok")
+
+
+def check_query():
+    """Check that the credentials has permission to query from the Bucket"""
+    print("> Checking credentials for query ...", end=" ")
+    try:
+        client.query_api().query(f"from(bucket:\"{bucket}\") |> range(start: -1m) |> limit(n:1)", org)
+    except ApiException as e:
+        # missing credentials
+        if e.status == 404:
+            raise Exception(f"The specified token doesn't have sufficient credentials to read from '{bucket}' "
+                            f"or specified bucket doesn't exists.") from e
+        raise
+    print("ok")
+
+
+def check_write():
+    """Check that the credentials has permission to write into the Bucket"""
+    print("> Checking credentials for write ...", end=" ")
+    try:
+        client.write_api(write_options=SYNCHRONOUS).write(bucket, org, b"")
+    except ApiException as e:
+        # bucket does not exist
+        if e.status == 404:
+            raise Exception(f"The specified bucket does not exist.") from e
+        # insufficient permissions
+        if e.status == 403:
+            raise Exception(f"The specified token does not have sufficient credentials to write to '{bucket}'.") from e
+        # 400 (BadRequest) caused by empty LineProtocol
+        if e.status != 400:
+            raise
+    print("ok")
+
+
+with InfluxDBClient(url=url, token=token, org=org) as client:
+    check_connection()
+    check_query()
+    check_write()
+    pass