Merge pull request #7 from GeographicaGS/feature/big-query-back

[FEATURE] Added Big Query backend

.env.example

@@ -15,14 +15,19 @@ GLUTEMULO_CARTO_USER=pcaro
 GLUTEMULO_CARTO_API_KEY=AADDAADDSS
 GLUTEMULO_CARTO_ORG=
 
-# when your backend in redis
+# when your backend is Redis
 GLUTEMULO_REDIS_HOST=redis
 #GLUTEMULO_REDIS_PASSWORD=redis_password
 GLUTEMULO_REDIS_PORT=6379
 #GLUTEMULO_REDIS_DB=0
 #GLUTEMULO_REDIS_KEY_PREFIX="gluto:"
 #GLUTEMULO_REDIS_EXPIRE_SECONDS=900
 
+# when your backend is Big Query
+GLUTEMULO_BQ_PROJECT=project
+GLUTEMULO_BQ_DATASET=gluto
+GLUTEMULO_BQ_TABLE=log
+
 GLUTEMULO_INGESTOR_TOPIC=simple-topic
 GLUTEMULO_INGESTOR_BOOTSTRAP_SERVERS=192.168.240.41:9092
 GLUTEMULO_INGESTOR_GROUP_ID=A group

glutemulo/backend/big_query.py

@@ -0,0 +1,17 @@
+from google.cloud import bigquery
+
+
+class BigQueryBackend:
+    def __init__(self, project, dataset, table, table_columns):
+        self.big_query = bigquery.Client(project)
+        table_ref = self.big_query.dataset(dataset).table(table)
+        self.table = self.big_query.get_table(table_ref)
+        self.columns = table_columns
+
+
+    def consume(self, messages):
+        data = [[msg.get(k, "") for k in self.columns] for msg in messages]
+        self.copy(data)
+
+    def copy(self, rows):
+        raise Exception('Method not implemented')

glutemulo/config.py

@@ -89,3 +89,12 @@
                     },
                 }
             )
+    elif backend == "big_query" and ingestor_enabled:
+        with env.prefixed("BQ_"):
+            config.update(
+                {
+                    "bq_project": env.str("PROJECT", "project"),
+                    "bq_dataset": env.str("DATASET", "gluto"),
+                    "bq_table": env.str("TABLE", "log"),
+                }
+            )

glutemulo/gluto.py

@@ -17,6 +17,10 @@
     from glutemulo.backend.redis import RedisBackend
 
     log.debug("Using REDIS backend")
+elif config["backend"] == "big_query":
+    from glutemulo.backend.big_query import BigQueryBackend
+
+    log.debug("Using Big Query backend")
 else:
     from glutemulo.backend.logger import LoggerBackend as Backend
 
@@ -46,6 +50,13 @@
             config["redis_key_prefix"],
             **config["redis_connection"],
         )
+    elif config["backend"] == "big_query":
+        backend = BigQueryBackend(
+            config["bq_project"],
+            config["bq_dataset"],
+            config["bq_table"],
+            []
+        )
     else:
 
         backend = Backend(

pyproject.toml

@@ -17,6 +17,7 @@ fastavro = "^0.21.23"
 carto = {version = "^1.4",optional = true}
 psycopg2-binary = {version = "^2.8",optional = true}
 redis = {version = "^3.2",optional = true}
+google-cloud-bigquery = "^1.21"
 
 [tool.poetry.dev-dependencies]
 flask = "^1.0"