Refacto/Modin to_sql (#201)

* wip

* wip: readme

* doc: delete where example

README.md

@@ -262,6 +262,12 @@ If our media perimeter evolves, we have to reimport it all using env variable `S
 
 Otherwise, default is yesterday midnight date (default cron job)
 
+**As pandas to_sql does not enable upsert (update/insert)**, if we want to update already saved rows, we have to delete first the rows and then start the program with `START_DATE` :
+```
+DELETE FROM keywords
+WHERE start BETWEEN '2024-05-01' AND '2024-05-30';
+```
+
 ### Based on channel
 Use env variable `CHANNEL` like in docker compose (string: tf1)
 

docker-compose.yml

@@ -137,7 +137,7 @@ services:
     #entrypoint: ["python", "quotaclimat/data_processing/mediatree/api_import.py"]
     environment:
       ENV: docker # change me to prod for real cases
-      LOGLEVEL: DEBUG # Change me to info (debug, info, warning, error) to have less log
+      LOGLEVEL: WARNING # Change me to info (debug, info, warning, error) to have less log
       PYTHONPATH: /app
       POSTGRES_USER: user
       POSTGRES_DB: barometre

postgres/schemas/models.py

@@ -66,6 +66,7 @@ class Keywords(Base):
     plaintext= Column(Text)
     theme=Column(JSON) #keyword.py  # ALTER TABLE keywords ALTER theme TYPE json USING to_json(theme);
     created_at = Column(DateTime(timezone=True), server_default=text("(now() at time zone 'utc')")) # ALTER TABLE ONLY keywords ALTER COLUMN created_at SET DEFAULT (now() at time zone 'utc');
+    updated_at = Column(DateTime(), default=datetime.now, onupdate=text("now() at time zone 'Europe/Paris'"), nullable=True)
     keywords_with_timestamp = Column(JSON) # ALTER TABLE keywords ADD keywords_with_timestamp json;
     number_of_keywords = Column(Integer) # ALTER TABLE keywords ADD number_of_keywords integer;
     srt = Column(JSON) # ALTER TABLE keywords ADD srt json;

quotaclimat/data_processing/mediatree/api_import.py

@@ -109,9 +109,8 @@ async def get_and_save_api_data(exit_event):
                             logging.info(f"Querying API for {channel} - {channel_program} - {channel_program_type} - {start_epoch} - {end_epoch}")
                             df = extract_api_sub(token, channel, type_sub, start_epoch,end_epoch, channel_program,channel_program_type) 
                             if(df is not None):
-                                # must ._to_pandas() because modin to_sql is not working
-                                save_to_pg(df._to_pandas(), keywords_table, conn)
-                            else: 
+                                save_to_pg(df, keywords_table, conn)
+                            else:
                                 logging.info("Nothing to save to Postgresql")
                     except Exception as err:
                         logging.error(f"continuing loop but met error : {err}")