splitting up text and detecting circular import

Author: HansVRP

openeo/extra/job_management/_job_db.py

@@ -203,42 +203,3 @@ def persist(self, df: pd.DataFrame):
         self.df.to_parquet(self.path, index=False)
 
 
-def get_job_db(path: Union[str, Path]) -> JobDatabaseInterface:
-    """
-    Factory to get a job database at a given path,
-    guessing the database type from filename extension.
-
-    :param path: path to job database file.
-
-    .. versionadded:: 0.33.0
-    """
-    path = Path(path)
-    if path.suffix.lower() in {".csv"}:
-        job_db = CsvJobDatabase(path=path)
-    elif path.suffix.lower() in {".parquet", ".geoparquet"}:
-        job_db = ParquetJobDatabase(path=path)
-    else:
-        raise ValueError(f"Could not guess job database type from {path!r}")
-    return job_db
-
-
-def create_job_db(path: Union[str, Path], df: pd.DataFrame, *, on_exists: str = "error"):
-    """
-    Factory to create a job database at given path,
-    initialized from a given dataframe,
-    and its database type guessed from filename extension.
-
-    :param path: Path to the job database file.
-    :param df: DataFrame to store in the job database.
-    :param on_exists: What to do when the job database already exists:
-        - "error": (default) raise an exception
-        - "skip": work with existing database, ignore given dataframe and skip any initialization
-
-    .. versionadded:: 0.33.0
-    """
-    job_db = get_job_db(path)
-    if isinstance(job_db, FullDataFrameJobDatabase):
-        job_db.initialize_from_df(df=df, on_exists=on_exists)
-    else:
-        raise NotImplementedError(f"Initialization of {type(job_db)} is not supported.")
-    return job_db

openeo/extra/job_management/_manager.py

@@ -31,6 +31,7 @@
     _JobStartTask,
 )
 from openeo.extra.job_management._interface import JobDatabaseInterface
+#from openeo.extra.job_management._job_db import get_job_db #TODO circular import
 
 from openeo.rest import OpenEoApiError
 from openeo.rest.auth.auth import BearerAuth
@@ -43,6 +44,8 @@
 # Sentinel value to indicate that a parameter was not set
 _UNSET = object()
 
+def _start_job_default(row: pd.Series, connection: Connection, *args, **kwargs):
+    raise NotImplementedError("No 'start_job' callable provided")
 
 class _Backend(NamedTuple):
     """Container for backend info/settings"""
@@ -434,7 +437,7 @@ def run_jobs(
         assert not kwargs, f"Unexpected keyword arguments: {kwargs!r}"
 
         if isinstance(job_db, (str, Path)):
-            job_db = get_job_db(path=job_db)
+            job_db = get_job_db(path=job_db) #TODO circular import
 
         if not isinstance(job_db, JobDatabaseInterface):
             raise ValueError(f"Unsupported job_db {job_db!r}")
@@ -832,8 +835,6 @@ def _track_statuses(self, job_db: JobDatabaseInterface, stats: Optional[dict] =
 
         return jobs_done, jobs_error, jobs_cancel
 
-def _start_job_default(row: pd.Series, connection: Connection, *args, **kwargs):
-    raise NotImplementedError("No 'start_job' callable provided")
 
 def _format_usage_stat(job_metadata: dict, field: str) -> str:
     value = deep_get(job_metadata, "usage", field, "value", default=0)