Merge pull request #58 from telefonicasc/fix/sql_replaceid

added replace_id attrib to sqlFileStore

Author: arcosa

python-lib/tc_etl_lib/README.md

@@ -341,6 +341,9 @@ La librería además proporciona [context managers](https://docs.python.org/3/re
         - si `table_name[entityType]` existe y es *falsy* (`None`, `""`, etc), las entidades de ese tipo no se escriben al fichero SQL.
     - :param: `chunk_size` opcional: máximo número de líneas a incluir en un solo `INSERT`. Default=10000
     - :param: `append` opcional: en caso de que el fichero exista, `append=True` añade los INSERT mientras que `append=False` sobrescribe el fichero. Default False.
+    - :param `replace_id` opcional: diccionario `tipo de entidad` => `lista de atributos replace_id`.
+        Reemplaza el ID de las entidades del tipo o tipos especificados, por un valor construido a partir de la lista de atributos indicados en este parámetro, separados por `_`.
+        Imita el comportamiento del atributo `replaceId` de los FLOW_HISTORIC de URBO DEPLOYER, para poder usar este *store* en ETLs que alimenten *singletons*.
     - :return: un `callable` que recibe una lista de entidades y las escribe como instrucciones sql `INSERT` en el fichero especificado.
 
 El modo de uso de cualquiera de los context managers es idéntico:
@@ -439,6 +442,8 @@ TOTAL                        403    221    45%
 
 ## Changelog
 
+- Add: new optional parameter called `replace_id` in sqlFileStore context manager ([#58](https://github.com/telefonicasc/etl-framework/pull/58))
+
 0.7.0 (December 233rd, 2022)
 
 - Add: new stores for saving entity batches, `orionStore` and `sqlFileStore` ([#46](https://github.com/telefonicasc/etl-framework/pull/46))

python-lib/tc_etl_lib/tc_etl_lib/store.py

@@ -23,7 +23,7 @@
   - orionStore: saves batches to Orion environment
   - sqlFileStore: saves batches to SQL File
 '''
-from typing import Callable, Dict, Any, Iterable, Sequence, Optional
+from typing import Callable, Dict, Any, Iterable, Sequence, List, Optional
 from contextlib import contextmanager
 from pathlib import Path
 from collections import defaultdict
@@ -48,7 +48,7 @@ def send_batch(entities: Iterable[Any]):
     yield send_batch
 
 @contextmanager
-def sqlFileStore(path: Path, *, subservice:str, schema:str=":target_schema", namespace:str="", table_names:Optional[Dict[str, str]]=None, chunk_size:int=10000, append:bool=False):
+def sqlFileStore(path: Path, *, subservice:str, schema:str=":target_schema", namespace:str="", table_names:Optional[Dict[str, str]]=None, chunk_size:int=10000, append:bool=False, replace_id:Optional[Dict[str, Sequence[str]]]=None):
     '''
     Context manager that creates a store to save entities to an SQL File.
     SQL syntax used is postgresql.
@@ -62,14 +62,20 @@ def sqlFileStore(path: Path, *, subservice:str, schema:str=":target_schema", nam
       - if table_name[entityType] exists and is empty, entities with the given type are not saved.
     chunk_size: maximum lines in a single insert statement. Default=10000
     append: append to the file instead of overwriting.
+    replace_id: dictionary of `entity type` => `list of replace_id attributes`.
+        For any entity of the given type, the entity id will be replaced by
+        the list of values of the specified attributes, concatenated with "_".
+        This mimics the behaviour of `replaceId` param in historic URBO-DEPLOYER flows.
     '''
     mode = "a+" if append else "w+"
     handler = path.open(mode=mode, encoding="utf-8")
     some_table_names = table_names or {} # make sure it is not None
+    replace_id = replace_id or {} # make sure it is not None
     try:
         def send_batch(entities: Iterable[Any]):
+            """Send a batch of entities to the database"""
             for chunk in iter_chunk(entities, chunk_size):
-                handler.write(sqlfile_batch(schema=schema, namespace=namespace, table_names=some_table_names, subservice=subservice, entities=chunk))
+                handler.write(sqlfile_batch(schema=schema, namespace=namespace, table_names=some_table_names, subservice=subservice, replace_id=replace_id, entities=chunk))
                 handler.write("\n")
         yield send_batch
     finally:
@@ -100,15 +106,26 @@ def sql_escape(obj: Any) -> str:
         adaptor.encoding = 'utf-8'
     return adaptor.getquoted().decode('utf-8')
 
-def sqlfile_values(subservice: str, entity: Dict[str, Any], fields: Iterable[str]) -> str:
+def sqlfile_values(subservice: str, entity: Dict[str, Any], fields: Iterable[str], replace_id:Optional[Sequence[str]]=None) -> str:
     '''
     Generates a string suitable for SQL insert, with all values of the entity
     
     subservice: subservice name
     entity: ngsi entity
-    fields: list of fields to save in the entity (omitting id, type)'''
+    fields: list of fields to save in the entity (omitting id, type)
+    replace_id: optional list of attributes to replace the entity's id
+    '''
+    entityid = entity['id']
+    if replace_id is not None and len(replace_id) > 0:
+        values = []
+        for attr in replace_id:
+            if attr == 'id':
+                values.append(entityid)
+            else:
+                values.append(str(entity[attr]['value']))
+        entityid = "_".join(values)
     sql = [
-        sql_escape(entity['id']),
+        sql_escape(entityid),
         sql_escape(entity['type']),
         sql_escape(subservice),
         "NOW()"
@@ -128,19 +145,20 @@ def sqlfile_values(subservice: str, entity: Dict[str, Any], fields: Iterable[str
         sql.append(value)
     return f"({','.join(sql)})"
 
-def sqlfile_insert(subservice: str, table_name: str, fields: Sequence[str], entities: Iterable[Any]) -> str:
+def sqlfile_insert(subservice: str, table_name: str, fields: Sequence[str], entities: Iterable[Any], replace_id:Optional[Sequence[str]]=None) -> str:
     '''
     Generate SQL INSERT lines from sequence of entities
     
     subservice: subservice name
     table_name: SQL table name to use
     fields: sequence of attribute names
     entities: iterable of entities
+    replace_id: optional list of alues to replace the entity's id
     '''
     return "\n".join((
         f"INSERT INTO {table_name} (entityid,entitytype,fiwareservicepath,recvtime,{','.join(fields)}) VALUES",
         ",\n".join(
-            sqlfile_values(subservice, entity, fields)
+            sqlfile_values(subservice, entity, fields, replace_id)
             for entity in entities
         ) + ";"
     ))
@@ -160,14 +178,15 @@ def sql_table_name(schema: str, namespace: str, entity_type: str, table_names: D
     # Tables mapped to some name are prefixed with schema name
     return f"{schema}.{mapped_name}"
 
-def sqlfile_batch(schema: str, namespace: str, table_names: Dict[str, str], subservice: str, entities: Iterable[Any]) -> str:
+def sqlfile_batch(schema: str, namespace: str, table_names: Dict[str, str], subservice: str, replace_id: Dict[str, Sequence[str]], entities: Iterable[Any]) -> str:
     '''
     Generate a single SQL insert batch statement
 
     schema: schema name
     namespace: namespace for table names
     table_names: overrides entytyType to table name defaults
     subservice: subservice name
+    replace_id: map of entity type to list of replace_id attributes
     entities: Iterable of entities
     '''
     # Group entities by type
@@ -187,6 +206,6 @@ def sqlfile_batch(schema: str, namespace: str, table_names: Dict[str, str], subs
         table_name = sql_table_name(schema, namespace, entity_type, table_names)
         if table_name:
             table_cols = sorted(fields_by_type[entity_type])
-            sql.append(sqlfile_insert(subservice, table_name, table_cols, typed_entities))
+            sql.append(sqlfile_insert(subservice, table_name, table_cols, typed_entities, replace_id.get(entity_type, None)))
     # and return SQL insert code
     return "\n".join(sql)

python-lib/tc_etl_lib/tc_etl_lib/test_store.py

@@ -100,7 +100,7 @@ def dedent(text: str) -> str:
 class TestSQLFileStore(unittest.TestCase):
     '''Tests for sqlFileStore'''
 
-    def do_test(self, expect: str, append_text: str, **params):
+    def do_test(self, expect: str, append_text: str, entities=TestEntities, **params):
         '''test sqlFileStore with given parameters and expectations'''
         with tempfile.NamedTemporaryFile(mode='w+', encoding='utf-8', delete=False) as tmpFile:
             try:
@@ -109,7 +109,7 @@ def do_test(self, expect: str, append_text: str, **params):
                     with open(tmpFile.name, "w+", encoding="utf-8") as outfile:
                         outfile.write(dedent(append_text))
                 with sqlFileStore(Path(tmpFile.name), **params) as store:
-                    store(TestEntities)
+                    store(entities)
                 with open(tmpFile.name, "r", encoding="utf-8") as infile:
                     data = infile.read()
                     self.maxDiff = None
@@ -213,3 +213,103 @@ def test_append(self):
         ('id_4','type_B','/testsrv',NOW(),'2022-12-15T18:01:00Z',21);
         """
         self.do_test(expected, create, subservice="/testsrv", schema="myschema", chunk_size=3, append=True)
+
+    def test_singleton_one_id(self):
+        '''Test replace_id with a single attribute'''
+        entities  = [
+            {
+                "id": "id_singleton",
+                "type": "type_A",
+                "ref": {
+                    "type": "Text",
+                    "value": "id_A"
+                },
+                "municipality": {
+                    "type": "Text",
+                    "value": "NA"
+                },
+                "location": {
+                    "type": "geo:json",
+                    "value": { "type": "Point", "coordinates": [1, 2] }
+                }
+            },
+            {
+                "id": "id_no_singleton",
+                "type": "type_B",
+                "ref": {
+                    "type": "Text",
+                    "value": "id_B"
+                },
+                "municipality": {
+                    "type": "Text",
+                    "value": "NA"
+                },
+                "location": {
+                    "type": "geo:json",
+                    "value": { "type": "Point", "coordinates": [1, 2] }
+                }
+            }
+        ]
+        expected = """
+        INSERT INTO myschema.type_a (entityid,entitytype,fiwareservicepath,recvtime,location,municipality,ref) VALUES
+        ('id_A','type_A','/testsrv',NOW(),ST_GeomFromGeoJSON('{"type": "Point", "coordinates": [1, 2]}'),'NA','id_A');
+        INSERT INTO myschema.type_b (entityid,entitytype,fiwareservicepath,recvtime,location,municipality,ref) VALUES
+        ('id_no_singleton','type_B','/testsrv',NOW(),ST_GeomFromGeoJSON('{"type": "Point", "coordinates": [1, 2]}'),'NA','id_B');
+        """
+        self.do_test(expected, "", entities=entities, replace_id={'type_A':['ref']}, subservice="/testsrv", schema="myschema")
+
+    def test_singleton_two_ids(self):
+        '''Test replace_id with two attributes'''
+        entities  = [
+            {
+                "id": "id_singleton",
+                "type": "type_A",
+                "primary": {
+                    "type": "Text",
+                    "value": "id_primary"
+                },
+                "secondary": {
+                    "type": "Number",
+                    "value": 5
+                },
+                "municipality": {
+                    "type": "Text",
+                    "value": "NA"
+                },
+                "location": {
+                    "type": "geo:json",
+                    "value": { "type": "Point", "coordinates": [1, 2] }
+                }
+            }
+        ]
+        expected = """
+        INSERT INTO myschema.type_a (entityid,entitytype,fiwareservicepath,recvtime,location,municipality,primary,secondary) VALUES
+        ('id_primary_5','type_A','/testsrv',NOW(),ST_GeomFromGeoJSON('{"type": "Point", "coordinates": [1, 2]}'),'NA','id_primary',5);
+        """
+        self.do_test(expected, "", entities=entities, replace_id={'type_A':['primary', 'secondary']}, subservice="/testsrv", schema="myschema")
+
+    def test_singleton_original_id(self):
+        '''Test replace_id with the original id, plus some attrib'''
+        entities  = [
+            {
+                "id": "id_singleton",
+                "type": "type_A",
+                "ref": {
+                    "type": "Number",
+                    "value": 5
+                },
+                "municipality": {
+                    "type": "Text",
+                    "value": "NA"
+                },
+                "location": {
+                    "type": "geo:json",
+                    "value": { "type": "Point", "coordinates": [1, 2] }
+                }
+            }
+        ]
+        expected = """
+        INSERT INTO myschema.type_a (entityid,entitytype,fiwareservicepath,recvtime,location,municipality,ref) VALUES
+        ('id_singleton_5','type_A','/testsrv',NOW(),ST_GeomFromGeoJSON('{"type": "Point", "coordinates": [1, 2]}'),'NA',5);
+        """
+        self.do_test(expected, "", entities=entities, replace_id={'type_A':['id', 'ref']}, subservice="/testsrv", schema="myschema")