fix(io): ensure that topic is created in PubSub writer

* Update documentation for PubSub & Kafka writers
* Add array handling, date handling, and prefix/suffix usage

Author: AndreyMarkinPPC

docs/usage/writers.md

@@ -92,7 +92,103 @@ concrete_writer.write(sample_report, 'query')
 
 ## Configuration
 
+### arrays
+
 Each of writer also support two options for dealing with arrays:
 
-* `WRITER.array-handling` - arrays handling method: "strings" (default)  - store arrays as strings (items combined via a separator, e.g. "item1|item2"), "arrays" - store arrays as arrays.
-* `WRITER.array-separator` - a separator symbol for joining arrays as strings, by default '|'.
+* `WRITER.array_handling` - arrays handling method:
+    * `strings` (default)  - store arrays as strings (items combined via a separator, e.g. "item1|item2")
+    * `arrays` - store arrays as arrays.
+* `WRITER.array_separator` - a separator symbol for joining arrays as strings, by default '|'.
+
+/// tab | cli
+```bash
+garf query.sql --source API_SOURCE \
+  --output json \
+  --json.array-handling=arrays
+
+garf query.sql --source API_SOURCE \
+  --output json \
+  --json.array-handling=strings --json.array-separator='*'
+```
+///
+
+/// tab | python
+```python
+from garf.io.writers import json_writer
+
+
+array_writer = json_writer.JsonWriter(array_handling='arrays')
+string_writer = json_writer.JsonWriter(
+  array_handling='strings', array_separator='*'
+)
+
+```
+
+### dates
+
+By default `garf` writes all date objects as strings. You can overwrite this with two options:
+
+* `WRITER.date_handling` - specifies ways of handling date object:
+    * `strings` (default)  - keeps date objects as strings .
+    * `date` - formats date objects to proper dates.
+    * `datetimes` - formats date objects to proper datetimes.
+    * `timestamps` - formats date objects to proper timestamps.
+* `WRITER.date_format_string` - specifies [format string](https://docs.python.org/3/library/datetime.html#format-codes).
+
+/// tab | cli
+```bash
+garf query.sql --source API_SOURCE \
+  --output bq \
+  --bq.date-handling=dates
+
+garf query.sql --source API_SOURCE \
+  --output bq \
+  --bq.date-handling=dates --json.date-format-string='%d/%m/%y'
+```
+///
+
+/// tab | python
+```python
+from garf.io.writers import bigquery_writer
+
+
+dates_writer = bigquery_writer.BigQueryWriter(date_handling='dates')
+dates_writer_with_format = bigquery_writer.BigQueryWriter(
+  date_handling='dates',
+  date_format_string='%d/%m/%y',
+)
+```
+
+### prefix / suffix
+
+When writing data with `garf` you can use `prefix` and `suffix` to dynamically
+update where (table / file / topic / index) data are written:
+
+/// tab | cli
+```bash
+# Saves results to `my_prefix_query.csv'
+garf query.sql --source API_SOURCE \
+  --output csv \
+  --csv.prefix=my_prefix
+
+# Saves results to `query_my_suffix.csv'
+garf query.sql --source API_SOURCE \
+  --output csv \
+  --csv.suffix=my_suffix
+
+# Saves results to `my_prefix_query_my_suffix.csv'
+garf query.sql --source API_SOURCE \
+  --output csv \
+  --csv.prefix=my_prefix \
+  --csv.suffix=my_suffix
+```
+///
+
+/// tab | python
+```python
+from garf.io.writers import csv_writer
+
+# Saves results to `my_prefix_query_my_suffix.csv'
+writer = csv_writer.CsvWriter(prefix='my_prefix', suffix='my_suffix')
+```

docs/usage/writers/kafka-writer.md

@@ -55,3 +55,58 @@ writer = kafka_writer.KafkaWriter(bootstrap_servers="broker1:9092,broker2:9092")
 writer.write(sample_report, 'topic_name')
 ```
 ///
+
+### push_strategy
+
+By default Kafka writer pushes the whole report in a body.
+You can overwrite it with `push_strategy` parameter which supports three options:
+
+  * `report` - pushes the whole report as a message.
+  * `batch` - pushes N rows of report into a message.
+  * `row` - pushes each row of report into a separate message.
+
+/// tab | cli
+```bash hl_lines="3"
+garf query.sql --source API_SOURCE \
+  --output kafka \
+  --kafka.push-strategy=row
+```
+///
+
+/// tab | python
+```python hl_lines="7"
+from garf.core import report
+from garf.io.writers import kafka_writer
+
+sample_report = report.GarfReport(results=[[1]], column_names=['one'])
+
+writer = kafka_writer.KafkaWriter(push_strategy='row')
+writer.write(sample_report, 'topic_name')
+```
+///
+
+### batch_size
+
+For `batch` `push_strategy` the default number of messages in a batch is `10`.
+You can overwrite it with `batch_size` parameter.
+
+/// tab | cli
+```bash hl_lines="3"
+garf query.sql --source API_SOURCE \
+  --output kafka \
+  --kafka.push-strategy=batch \
+  --kafka.batch-size=5
+```
+///
+
+/// tab | python
+```python hl_lines="7"
+from garf.core import report
+from garf.io.writers import kafka_writer
+
+sample_report = report.GarfReport(results=[[1]], column_names=['one'])
+
+writer = kafka_writer.KafkaWriter(push_strategy='batch', batch_size=5)
+writer.write(sample_report, 'topic_name')
+```
+///

docs/usage/writers/pubsub-writer.md

@@ -30,7 +30,7 @@ writer.write(sample_report, 'topic_name')
 ///
 
 ## Parameters
-### Project
+### project
 
 By default it uses `GOOGLE_CLOUD_PROJECT` environment variable.
 You can overwrite it with `project` parameter.
@@ -48,10 +48,64 @@ garf query.sql --source API_SOURCE \
 from garf.core import report
 from garf.io.writers import pubsub_writer
 
-# Create example report
 sample_report = report.GarfReport(results=[[1]], column_names=['one'])
 
 writer = pubsub_writer.PubSubWriter(project="ANOTHER_PROJECT_ID")
 writer.write(sample_report, 'topic_name')
 ```
 ///
+
+### push_strategy
+
+By default PubSub writer pushes the whole report in a body.
+You can overwrite it with `push_strategy` parameter which supports three options:
+
+  * `report` - pushes the whole report as a message.
+  * `batch` - pushes N rows of report into a message.
+  * `row` - pushes each row of report into a separate message.
+
+/// tab | cli
+```bash hl_lines="3"
+garf query.sql --source API_SOURCE \
+  --output pubsub \
+  --pubsub.push-strategy=row
+```
+///
+
+/// tab | python
+```python hl_lines="7"
+from garf.core import report
+from garf.io.writers import pubsub_writer
+
+sample_report = report.GarfReport(results=[[1]], column_names=['one'])
+
+writer = pubsub_writer.PubSubWriter(push_strategy='row')
+writer.write(sample_report, 'topic_name')
+```
+///
+
+### batch_size
+
+For `batch` `push_strategy` the default number of messages in a batch is `10`.
+You can overwrite it with `batch_size` parameter.
+
+/// tab | cli
+```bash hl_lines="3"
+garf query.sql --source API_SOURCE \
+  --output pubsub \
+  --pubsub.push-strategy=batch \
+  --pubsub.batch-size=5
+```
+///
+
+/// tab | python
+```python hl_lines="7"
+from garf.core import report
+from garf.io.writers import pubsub_writer
+
+sample_report = report.GarfReport(results=[[1]], column_names=['one'])
+
+writer = pubsub_writer.PubSubWriter(push_strategy='batch', batch_size=5)
+writer.write(sample_report, 'topic_name')
+```
+///

libs/io/garf/io/__init__.py

@@ -13,4 +13,4 @@
 # limitations under the License.
 """Write GarfReport to anywhere."""
 
-__version__ = '1.3.0'
+__version__ = '1.3.1'

libs/io/garf/io/writers/pubsub_writer.py

@@ -25,6 +25,8 @@
     'Please install garf-io with PubSub support - `pip install garf-io[pubsub]`'
   ) from e
 
+from google.api_core.exceptions import NotFound
+
 logger = logging.getLogger(__name__)
 
 
@@ -56,8 +58,13 @@ def _init_producer(self):
 
   def create_topic(self, topic: str) -> str:
     topic_path = self.publisher.topic_path(self.project, topic)
-    if not self.publisher.get_topic(request={'topic': topic_path}):
-      self.publisher.create_topic(request={'name': topic_path})
+    request = {'name': topic_path}
+    try:
+      if not self.publisher.get_topic(request={'topic': topic_path}):
+        self.publisher.create_topic(request=request)
+    except NotFound:
+      self.publisher.create_topic(request=request)
+
     return topic_path
 
   def _send(self, data: bytes, topic: str) -> None:

libs/io/garf/io/writers/topic_writer.py

@@ -46,7 +46,7 @@ def __init__(
     batch_size: int = 10,
     **kwargs: str,
   ) -> None:
-    """Initializes KafkaWriter."""
+    """Initializes TopicWriter."""
     super().__init__(**kwargs)
     self.provider = provider
     self.push_strategy = push_strategy
@@ -62,14 +62,14 @@ def _init_producer(self):
     raise NotImplementedError
 
   def write(self, report: garf_report.GarfReport, destination: str) -> str:
-    """Writes report to Kafka topic.
+    """Writes report to topic.
 
     Args:
       report: GarfReport to write.
-      destination: Kafka topic name.
+      destination: Topic name.
     """
     self._init_producer()
-    with tracer.start_as_current_span('f{self.provider}.write') as span:
+    with tracer.start_as_current_span(f'{self.provider}.write') as span:
       span.set_attribute('writer.type', str(self.push_strategy))
       destination = formatter.format_extension(
         destination,