Changeset View
Standalone View
swh/clearlydefined/orchestrator.py
- This file was added.
| # Copyright (C) 2021 The Software Heritage developers | |||||
| # See the AUTHORS file at the top-level directory of this distribution | |||||
| # License: GNU Affero General Public License version 3, or any later version | |||||
| # See top-level LICENSE file for more information | |||||
| from datetime import datetime | |||||
| from typing import Optional | |||||
| import psycopg2 | |||||
| from swh.clearlydefined.mapping_utils import ( | |||||
| authority, | |||||
| fetcher, | |||||
| get_type_of_tool, | |||||
| map_row, | |||||
| ) | |||||
| from swh.model.model import RawExtrinsicMetadata | |||||
| from swh.storage.interface import StorageInterface | |||||
| def write_in_storage( | |||||
| storage: StorageInterface, | |||||
| metadata: RawExtrinsicMetadata, | |||||
| ) -> None: | |||||
| """ | |||||
| Take storage and metadata as input | |||||
| and add metadata in storage | |||||
| """ | |||||
| storage.raw_extrinsic_metadata_add([metadata]) | |||||
| def init_tables(cursor, connection) -> None: | |||||
| """ | |||||
| Take connection and cursor as input and initialize tables if | |||||
| they don't exists | |||||
| """ | |||||
| cursor.execute( | |||||
| """ | |||||
| CREATE TABLE IF NOT EXISTS unmapped_data (path VARCHAR PRIMARY KEY); | |||||
| CREATE EXTENSION IF NOT EXISTS hstore; | |||||
| CREATE TABLE IF NOT EXISTS last_run_date (attr hstore);""" | |||||
| ) | |||||
| connection.commit() | |||||
vlorentz: this belongs in the mapping, not the orchestrator. | |||||
| def init_storage(storage: StorageInterface) -> None: | |||||
| """ | |||||
| Take storage as input and add MetadataFetcher, MetadataAuthority inside storage | |||||
| """ | |||||
| storage.metadata_authority_add([authority]) | |||||
| storage.metadata_fetcher_add([fetcher]) | |||||
Done Inline Actionsin the sql/ dir. and you shouldn't need ON CONFLICT vlorentz: in the sql/ dir.
and you shouldn't need `ON CONFLICT` | |||||
Done Inline ActionsCan you elaborate on this ? TG1999: Can you elaborate on this ? | |||||
Done Inline Actionsall the initialization should be in sql/30-schema.sql, like swh-storage does. vlorentz: all the initialization should be in `sql/30-schema.sql`, like swh-storage does. | |||||
Done Inline ActionsOkay, then can I remove init_tables function? TG1999: Okay, then can I remove init_tables function? | |||||
| def write_next_date( | |||||
| cursor, update_connection, previous_date: Optional[datetime], new_date: datetime | |||||
| ) -> None: | |||||
| """ | |||||
| Take cursor, update_connection, previous_date, new_date as input | |||||
| and if it previous_date is None, then enter new_date, else | |||||
| update the date stored in table with new_date | |||||
| """ | |||||
| value = f'"date"=>"{new_date}"' | |||||
| if not previous_date: | |||||
| cursor.execute("""INSERT into last_run_date (attr )VALUES(%s)""", (value,)) | |||||
| else: | |||||
| cursor.execute( | |||||
| """UPDATE last_run_date SET attr = attr || %s :: hstore""", | |||||
Done Inline Actionsuse constants instead of functions. vlorentz: use constants instead of functions. | |||||
| (value,), | |||||
| ) | |||||
| update_connection.commit() | |||||
| def get_last_run_date(cursor) -> Optional[datetime]: | |||||
| """ | |||||
| Take cursor as input and get last run date from which | |||||
| new rows will be orchestered, return None if it's first | |||||
| orchestration | |||||
| """ | |||||
| cursor.execute("SELECT attr -> 'date' AS date FROM last_run_date;") | |||||
Done Inline Actionswe already have a way to initialize databases in a standard way across SWH. It also includes a "dbversion" field so we can keep track of migrations. Check out swh/storage/sql and sql/upgrades/ in swh-storage, and swh/core/db/ in swh-core. vlorentz: we already have a way to initialize databases in a standard way across SWH. It also includes a… | |||||
| rows = cursor.fetchall() | |||||
| if len(rows) < 1: | |||||
| return None | |||||
Done Inline Actionsmissing type annotation (StorageInterface) vlorentz: missing type annotation (`StorageInterface`) | |||||
| date = rows[0][0] | |||||
| return date | |||||
| def orchestor_row(storage: StorageInterface, cursor, connection, row: tuple) -> bool: | |||||
Done Inline ActionsIn what state? And what is the returned value? vlorentz: In what state?
And what is the returned value? | |||||
| """ | |||||
| Take storage, cursor, connection, row as input | |||||
| and if able to completely map that row then write | |||||
| data in storage, else store the ID in state | |||||
| """ | |||||
Done Inline Actionsmissing type annotations. vlorentz: missing type annotations. | |||||
| mapped = map_row(metadata=row[1], id=row[0], date=row[2], storage=storage) | |||||
| if not mapped: | |||||
| write_in_not_mapped(cd_path=row[0], cursor=cursor, write_connection=connection) | |||||
| return False | |||||
| else: | |||||
| mapping_status, data_list = mapped | |||||
| if not mapping_status: | |||||
Done Inline Actionswhat if there is already a line in last_run_date? vlorentz: what if there is already a line in `last_run_date`? | |||||
Done Inline Actionsprevious_date ensures that, if we have don't have a previous date, then only we will insert value TG1999: previous_date ensures that, if we have don't have a previous date, then only we will insert… | |||||
| write_in_not_mapped( | |||||
| cd_path=row[0], cursor=cursor, write_connection=connection | |||||
| ) | |||||
| for data in data_list: | |||||
| write_in_storage(storage=storage, metadata=data) | |||||
| return mapping_status | |||||
Done Inline ActionsThis is neither a good signature or a good docstring. Readers have no idea what row should be. It's also very easy to pass the wrong fields in row or pass them in the wrong order Instead, either use a class, or use a dict and document every field. The name also isn't very explicit; there should at least be a verb or a noun representing an action in function names, because functions do something. orchestor_row sounds like a variable or class name, because it does not describe an action/computation vlorentz: This is neither a good signature or a good docstring. Readers have no idea what `row` should be. | |||||
| def map_previously_unmapped_data(storage: StorageInterface, cursor, connection) -> None: | |||||
Done Inline Actionsinstead of a table for just one value; you could create a table that will hold this kind of data; for example with a "key" and a value" column. For now it will only have one row, but we may need to add more in the future. vlorentz: instead of a table for just one value; you could create a table that will hold this kind of… | |||||
| """ | |||||
| Take storage, cursor, connection as input and map previously | |||||
| unmapped data | |||||
| """ | |||||
| cursor.execute("SELECT path FROM unmapped_data ;") | |||||
| rows = cursor.fetchall() | |||||
| for row in rows: | |||||
| cd_path = row[0] | |||||
| cursor.execute("SELECT path,content,last_modified_date FROM clearcode_cditem WHERE path=%s;", (cd_path,)) | |||||
| unmapped_row = cursor.fetchall()[0] | |||||
| if orchestor_row( | |||||
| storage=storage, | |||||
| row=unmapped_row, | |||||
Done Inline Actionswhat if there is more than one? vlorentz: what if there is more than one? | |||||
| cursor=cursor, | |||||
| connection=connection, | |||||
| ): | |||||
| cursor.execute("DELETE FROM unmapped_data WHERE path=%s", (cd_path,)) | |||||
| connection.commit() | |||||
Done Inline Actionswhat? vlorentz: what? | |||||
| def write_in_not_mapped(cursor, write_connection, cd_path: str) -> None: | |||||
| """ | |||||
| Take cursor, write_connection, cd_path as input | |||||
| and write 'cd_path' if 'cd_path' does not exists | |||||
| inside unmapped_data | |||||
Done Inline ActionsIf I make this code run unconditionally, tests still pass. So either the conditional is useless or there is a test missing vlorentz: If I make this code run unconditionally, tests still pass. So either the conditional is useless… | |||||
Done Inline ActionsHey, your observation is correct, can you tell me how you inferred that, so I can correct this part of the code, since I have asserted orchestrate_row it also returns False (when needed), and I thought line 169in test_orchestrator, was covering this part, since we had 1 row in unmapped_data before running orchestrator again on 168, and it left 1 row after, so its not getting deleted as well, I am a little bit confused TG1999: Hey, your observation is correct, can you tell me how you inferred that, so I can correct this… | |||||
| """ | |||||
| cursor.execute( | |||||
| "INSERT INTO unmapped_data (path) VALUES (%s) ON CONFLICT (path) DO NOTHING;", | |||||
| (cd_path,), | |||||
| ) | |||||
| write_connection.commit() | |||||
| return | |||||
Done Inline ActionsI don't understand what is happening here. vlorentz: I don't understand what is happening here. | |||||
Done Inline ActionsI still don't. I don't see a change in the code and there are still no comments. (note: this comment applies to the second branch of orchestor_row) vlorentz: I still don't. I don't see a change in the code and there are still no comments. (note: this… | |||||
Done Inline ActionsI have updated the doc strings, if any other change is required can you elaborate on that? TG1999: I have updated the doc strings, if any other change is required can you elaborate on that? | |||||
Done Inline ActionsMostly, what is the content of mapped/data_list? It's rather confusing that the variable mapped seems to be related to the row that is not yet mapped. vlorentz: Mostly, what is the content of `mapped`/`data_list`?
It's rather confusing that the variable… | |||||
| def read_from_clearcode_and_write_in_swh( | |||||
| storage: StorageInterface, cursor, connection, date: Optional[datetime] | |||||
| ) -> None: | |||||
| """ | |||||
| Take storage, cursor, connection, date as input | |||||
| and read from clearcode database and write only | |||||
| the data that is discovered after 'date' in swh storage. | |||||
| 'date' is the last discovery date of the object that was | |||||
| stored at the time of previous run. | |||||
| """ | |||||
| if date: | |||||
| cursor.execute( | |||||
| "SELECT path,content,last_modified_date FROM clearcode_cditem " | |||||
| "WHERE last_modified_date < %s " | |||||
| "ORDER BY last_modified_date DESC;", | |||||
| (date,), | |||||
| ) | |||||
| else: | |||||
| cursor.execute( | |||||
Done Inline ActionsIf I understand this correct, your code first marks a row as mapped, then actually maps it. This is incorrect, because if the process crash between the two steps, the row will be incorrectly considered mapped. vlorentz: If I understand this correct, your code first marks a row as mapped, then actually maps it. | |||||
| "SELECT path,content,last_modified_date FROM clearcode_cditem ORDER BY last_modified_date DESC;" | |||||
| ) | |||||
| rows = cursor.fetchall() | |||||
| if len(rows) < 1: | |||||
| return | |||||
| new_date = rows[0][2] | |||||
| write_next_date( | |||||
| cursor=cursor, | |||||
| update_connection=connection, | |||||
| previous_date=date, | |||||
| new_date=new_date, | |||||
| ) | |||||
Done Inline Actionsuse an upsert instead vlorentz: use an upsert instead | |||||
| for row in rows: | |||||
| tool = get_type_of_tool(row[0]).value | |||||
| if tool == "fossology": | |||||
| pass | |||||
| else: | |||||
| orchestor_row( | |||||
| storage=storage, cursor=cursor, connection=connection, row=row | |||||
| ) | |||||
| def orchestrator(storage: StorageInterface, clearcode_dsn: str) -> None: | |||||
| """ | |||||
| Take clearcode_dsn, swh_storage_backend_config as input | |||||
| and write data periodically from clearcode database to | |||||
| swh raw extrensic metadata | |||||
| """ | |||||
| connection = psycopg2.connect(dsn=clearcode_dsn) | |||||
| cursor = connection.cursor() | |||||
| init_tables(cursor=cursor, connection=connection) | |||||
| init_storage(storage=storage) | |||||
| map_previously_unmapped_data(storage=storage, cursor=cursor, connection=connection) | |||||
| date = get_last_run_date(cursor=cursor) | |||||
| read_from_clearcode_and_write_in_swh( | |||||
| storage=storage, cursor=cursor, connection=connection, date=date | |||||
| ) | |||||
Done Inline Actionssame issue as above regarding the order vlorentz: same issue as above regarding the order | |||||
this belongs in the mapping, not the orchestrator.