" "Jobflow-Remote is still in beta phase. The API may change at any time." "
" @@ -137,14 +137,14 @@ # -- Options for HTMLHelp output --------------------------------------------- html_css_files = ["jobflow_remote.css"] -html_title = "%s v%s Manual" % (project, version) -html_last_updated_fmt = '%b %d, %Y' +html_title = f"{project} v{version} Manual" +html_last_updated_fmt = "%b %d, %Y" # html_css_files = ["numpy.css"] html_context = {"default_mode": "light"} html_use_modindex = True html_copy_source = False html_domain_indices = False -html_file_suffix = '.html' +html_file_suffix = ".html" # Output file base name for HTML help builder. htmlhelp_basename = "jobflow_remote_doc" diff --git a/doc/source/dev/index.rst b/doc/source/dev/index.rst index 88e18c08..4b7eca16 100644 --- a/doc/source/dev/index.rst +++ b/doc/source/dev/index.rst @@ -4,4 +4,4 @@ Contributing to Jobflow-Remote ############################## -Here are the things that can be done. \ No newline at end of file +Here are the things that can be done. diff --git a/doc/source/glossary.rst b/doc/source/glossary.rst index eff30732..bbdfcb41 100644 --- a/doc/source/glossary.rst +++ b/doc/source/glossary.rst @@ -11,4 +11,3 @@ Glossary Worker The description of a given resource where to execute flows. - diff --git a/doc/source/index.rst b/doc/source/index.rst index 609ce301..e4a05105 100644 --- a/doc/source/index.rst +++ b/doc/source/index.rst @@ -102,4 +102,4 @@ Jobflow-Remote is a package to submit jobflow flows remotely. .. This is not really the index page, that is found in _templates/indexcontent.html The toctree content here will be added to the - top of the template header \ No newline at end of file + top of the template header diff --git a/doc/source/user/basics.rst b/doc/source/user/basics.rst index bfaa0b3c..b76a13bd 100644 --- a/doc/source/user/basics.rst +++ b/doc/source/user/basics.rst @@ -9,4 +9,3 @@ fundamental Jobflow-Remote ideas and philosophy. .. .. toctree:: :maxdepth: 1 - diff --git a/doc/source/user/building.rst b/doc/source/user/building.rst index 93af682e..27837520 100644 --- a/doc/source/user/building.rst +++ b/doc/source/user/building.rst @@ -4,4 +4,4 @@ Building from source ==================== Get the source from the git repository. -Install it with pip install . \ No newline at end of file +Install it with pip install . diff --git a/doc/source/user/whatisjobflowremote.rst b/doc/source/user/whatisjobflowremote.rst index 093dee2e..772bf54d 100644 --- a/doc/source/user/whatisjobflowremote.rst +++ b/doc/source/user/whatisjobflowremote.rst @@ -5,4 +5,4 @@ What is Jobflow-Remote? ======================= Jobflow-Remote is ... -TODO: add the features that it has. \ No newline at end of file +TODO: add the features that it has. diff --git a/pyproject.toml b/pyproject.toml index 71fb42ab..3d9bd4bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,6 @@ authors = [{ name = "Guido Petretto", email = "guido.petretto@matgenix.com" }] dynamic = ["version"] classifiers = [ "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", @@ -24,7 +23,7 @@ classifiers = [ "Topic :: Other/Nonlisted Topic", "Topic :: Scientific/Engineering", ] -requires-python = ">=3.8" +requires-python = ">=3.9" dependencies =[ "jobflow[strict]", "pydantic<2", @@ -81,7 +80,7 @@ max-line-length = 88 max-doc-length = 88 select = "C, E, F, W, B" extend-ignore = "E203, W503, E501, F401, RST21" -min-python-version = "3.8.0" +min-python-version = "3.9.0" docstring-convention = "numpy" rst-roles = "class, func, ref, obj" diff --git a/src/jobflow_remote/cli/flow.py b/src/jobflow_remote/cli/flow.py index a13a2a6e..3ba4b0f6 100644 --- a/src/jobflow_remote/cli/flow.py +++ b/src/jobflow_remote/cli/flow.py @@ -171,7 +171,6 @@ def flow_info( flow_ids = [jf_id] with loading_spinner(): - jc = JobController() flows_info = jc.get_flows_info( diff --git a/src/jobflow_remote/cli/formatting.py b/src/jobflow_remote/cli/formatting.py index 7f942481..2b9fcf72 100644 --- a/src/jobflow_remote/cli/formatting.py +++ b/src/jobflow_remote/cli/formatting.py @@ -142,7 +142,6 @@ def format_job_info(job_info: JobInfo, show_none: bool = False): def format_flow_info(flow_info: FlowInfo): - title = f"Flow: {flow_info.name} - {flow_info.flow_id} - {flow_info.state.name}" table = Table(title=title) table.title_style = "bold" diff --git a/src/jobflow_remote/cli/job.py b/src/jobflow_remote/cli/job.py index 07f35780..9b215db1 100644 --- a/src/jobflow_remote/cli/job.py +++ b/src/jobflow_remote/cli/job.py @@ -159,7 +159,6 @@ def job_info( db_id, job_id = get_job_db_ids(job_db_id, job_index) with loading_spinner(): - jc = JobController() job_info = jc.get_job_info( diff --git a/src/jobflow_remote/cli/utils.py b/src/jobflow_remote/cli/utils.py index 355d0a03..ca250628 100644 --- a/src/jobflow_remote/cli/utils.py +++ b/src/jobflow_remote/cli/utils.py @@ -205,7 +205,6 @@ def convert_metadata(string_metadata: str | None) -> dict | None: def get_start_date(start_date: datetime | None, days: int | None, hours: int | None): - if start_date and (start_date.year, start_date.month, start_date.day) == ( 1900, 1, diff --git a/src/jobflow_remote/config/base.py b/src/jobflow_remote/config/base.py index 0317a61c..c0818e70 100644 --- a/src/jobflow_remote/config/base.py +++ b/src/jobflow_remote/config/base.py @@ -1,11 +1,9 @@ -from __future__ import annotations - import abc import logging import traceback from enum import Enum from pathlib import Path -from typing import Annotated, Literal +from typing import Annotated, Literal, Optional, Union from jobflow import JobStore from pydantic import BaseModel, Extra, Field, validator @@ -36,7 +34,7 @@ class RunnerOptions(BaseModel): 30, description="Delay between subsequent advancement of the job's remote state (seconds)", ) - lock_timeout: int | None = Field( + lock_timeout: Optional[int] = Field( 86400, description="Time to consider the lock on a document expired and can be overridden (seconds)", ) @@ -115,16 +113,16 @@ class WorkerBase(BaseModel): description="Absolute path of the directory of the worker where subfolders for " "executing the calculation will be created" ) - resources: dict | None = Field( + resources: Optional[dict] = Field( None, description="A dictionary defining the default resources requested to the " "scheduler. Used to fill in the QToolKit template", ) - pre_run: str | None = Field( + pre_run: Optional[str] = Field( None, description="String with commands that will be executed before the execution of the Job", ) - post_run: str | None = Field( + post_run: Optional[str] = Field( None, description="String with commands that will be executed after the execution of the Job", ) @@ -234,13 +232,13 @@ class RemoteWorker(WorkerBase): host: str = Field(description="The host to which to connect") user: str = Field(None, description="Login username") port: int = Field(None, description="Port number") - password: str | None = Field(None, description="Login password") - key_filename: str | list[str] | None = Field( + password: Optional[str] = Field(None, description="Login password") + key_filename: Optional[Union[str, list[str]]] = Field( None, description="The filename, or list of filenames, of optional private key(s) " "and/or certs to try for authentication", ) - passphrase: str | None = Field( + passphrase: Optional[str] = Field( None, description="Passphrase used for decrypting private keys" ) gateway: str = Field( @@ -259,10 +257,10 @@ class RemoteWorker(WorkerBase): description="Whether to send environment variables 'inline' as prefixes in " "front of command strings", ) - keepalive: int | None = Field( + keepalive: Optional[int] = Field( 60, description="Keepalive value in seconds passed to paramiko's transport" ) - shell_cmd: str | None = Field( + shell_cmd: Optional[str] = Field( "bash", description="The shell command used to execute the command remotely. If None " "the command is executed directly", @@ -318,7 +316,7 @@ def cli_info(self) -> dict: ) -WorkerConfig = Annotated[LocalWorker | RemoteWorker, Field(discriminator="type")] +WorkerConfig = Annotated[Union[LocalWorker, RemoteWorker], Field(discriminator="type")] class ExecutionConfig(BaseModel): @@ -326,14 +324,16 @@ class ExecutionConfig(BaseModel): Configuration to be set before and after the execution of a Job. """ - modules: list[str] | None = Field(None, description="list of modules to be loaded") - export: dict[str, str] | None = Field( + modules: Optional[list[str]] = Field( + None, description="list of modules to be loaded" + ) + export: Optional[dict[str, str]] = Field( None, description="dictionary with variable to be exported" ) - pre_run: str | None = Field( + pre_run: Optional[str] = Field( None, description="Other commands to be executed before the execution of a job" ) - post_run: str | None = Field( + post_run: Optional[str] = Field( None, description="Commands to be executed after the execution of a job" ) @@ -347,20 +347,20 @@ class Project(BaseModel): """ name: str = Field(description="The name of the project") - base_dir: str | None = Field( + base_dir: Optional[str] = Field( None, description="The base directory containing the project related files. Default " "is a folder with the project name inside the projects folder", ) - tmp_dir: str | None = Field( + tmp_dir: Optional[str] = Field( None, description="Folder where remote files are copied. Default a 'tmp' folder in base_dir", ) - log_dir: str | None = Field( + log_dir: Optional[str] = Field( None, description="Folder containing all the logs. Default a 'log' folder in base_dir", ) - daemon_dir: str | None = Field( + daemon_dir: Optional[str] = Field( None, description="Folder containing daemon related files. Default to a 'daemon' " "folder in base_dir", @@ -390,11 +390,11 @@ class Project(BaseModel): description="The JobStore used for the input. Can contain the monty " "serialized dictionary or the Store int the Jobflow format", ) - metadata: dict | None = Field( + metadata: Optional[dict] = Field( None, description="A dictionary with metadata associated to the project" ) - def get_jobstore(self) -> JobStore | None: + def get_jobstore(self) -> Optional[JobStore]: """ Generate an instance of the JobStore based on the configuration diff --git a/src/jobflow_remote/config/helper.py b/src/jobflow_remote/config/helper.py index a7e24c9a..063e1af5 100644 --- a/src/jobflow_remote/config/helper.py +++ b/src/jobflow_remote/config/helper.py @@ -2,7 +2,6 @@ import logging import traceback -from pathlib import Path from jobflow import JobStore from maggma.core import Store @@ -134,6 +133,7 @@ def _check_workdir(worker: WorkerBase, host: BaseHost) -> str | None: try: canary_file = worker.work_dir / ".jf_heartbeat" host.write_text_file(canary_file, "\n") + return None except FileNotFoundError as exc: raise FileNotFoundError( f"Could not write to {canary_file} on {worker.host}. Does the folder exist on the remote?\nThe folder should be specified as an absolute path with no shell expansions or environment variables." diff --git a/src/jobflow_remote/config/settings.py b/src/jobflow_remote/config/settings.py index 4b0610b1..82096b1c 100644 --- a/src/jobflow_remote/config/settings.py +++ b/src/jobflow_remote/config/settings.py @@ -10,7 +10,6 @@ class JobflowRemoteSettings(BaseSettings): - config_file: str = Field( DEFAULT_CONFIG_FILE_PATH, description="Location of the config file for jobflow remote.", diff --git a/src/jobflow_remote/fireworks/launcher.py b/src/jobflow_remote/fireworks/launcher.py index 0da92980..a666b0ec 100644 --- a/src/jobflow_remote/fireworks/launcher.py +++ b/src/jobflow_remote/fireworks/launcher.py @@ -32,7 +32,6 @@ def checkout_remote( launch_id = None try: - fw, launch_id = rlpad.lpad.reserve_fw(fworker, ".", fw_id=fw_id) if not fw: logger.info("No jobs exist in the LaunchPad for submission to queue!") diff --git a/src/jobflow_remote/fireworks/launchpad.py b/src/jobflow_remote/fireworks/launchpad.py index 2ae68c9a..8536ae1e 100644 --- a/src/jobflow_remote/fireworks/launchpad.py +++ b/src/jobflow_remote/fireworks/launchpad.py @@ -610,7 +610,6 @@ def get_remote_run( job_id: str | None = None, job_index: int | None = None, ) -> RemoteRun: - query, sort = self.generate_id_query(fw_id, job_id, job_index) fw = self.fireworks.find_one(query) @@ -697,7 +696,6 @@ def get_wf_fw_data( sort: dict | None = None, limit: int = 0, ) -> list[dict]: - pipeline: list[dict] = [ { "$lookup": { @@ -767,7 +765,6 @@ def get_fw_launch_remote_run_data( sort: dict | None = None, limit: int = 0, ) -> list[dict]: - # only take the most recent launch pipeline: list[dict] = [ { diff --git a/src/jobflow_remote/jobs/daemon.py b/src/jobflow_remote/jobs/daemon.py index 7753812a..32c1e6eb 100644 --- a/src/jobflow_remote/jobs/daemon.py +++ b/src/jobflow_remote/jobs/daemon.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import logging import subprocess from enum import Enum @@ -56,7 +58,6 @@ class DaemonStatus(Enum): class DaemonManager: - conf_template = Template(supervisord_conf_str) def __init__( diff --git a/src/jobflow_remote/jobs/jobcontroller.py b/src/jobflow_remote/jobs/jobcontroller.py index 3740092e..8596a373 100644 --- a/src/jobflow_remote/jobs/jobcontroller.py +++ b/src/jobflow_remote/jobs/jobcontroller.py @@ -147,7 +147,6 @@ def _build_query_wf( end_date: datetime | None = None, name: str | None = None, ) -> dict: - if job_ids is not None and not isinstance(job_ids, (list, tuple)): job_ids = [job_ids] if db_ids is not None and not isinstance(db_ids, (list, tuple)): @@ -391,7 +390,6 @@ def rerun_jobs( return fw_ids def reset(self, reset_output: bool = False, max_limit: int = 25) -> bool: - password = datetime.now().strftime("%Y-%m-%d") if max_limit == 0 else None try: self.rlpad.reset( diff --git a/src/jobflow_remote/jobs/runner.py b/src/jobflow_remote/jobs/runner.py index df22c082..f5cdcddd 100644 --- a/src/jobflow_remote/jobs/runner.py +++ b/src/jobflow_remote/jobs/runner.py @@ -240,11 +240,11 @@ def lock_and_update( error, fail_now, set_output = function(doc) except ConfigError: error = traceback.format_exc() - warnings.warn(error) + warnings.warn(error, stacklevel=2) fail_now = True except Exception: error = traceback.format_exc() - warnings.warn(error) + warnings.warn(error, stacklevel=2) lock.update_on_release = self._prepare_lock_update( doc, error, fail_now, set_output, state.next diff --git a/src/jobflow_remote/remote/data.py b/src/jobflow_remote/remote/data.py index 65d5e53d..9945a259 100644 --- a/src/jobflow_remote/remote/data.py +++ b/src/jobflow_remote/remote/data.py @@ -76,7 +76,6 @@ def get_remote_store_filenames(store: JobStore) -> list[str]: def update_store(store, remote_store, save): - # TODO is it correct? data = list(remote_store.query(load=save)) if len(data) > 1: diff --git a/src/jobflow_remote/utils/db.py b/src/jobflow_remote/utils/db.py index cc1a46ca..c1025dbc 100644 --- a/src/jobflow_remote/utils/db.py +++ b/src/jobflow_remote/utils/db.py @@ -12,7 +12,6 @@ class MongoLock: - LOCK_KEY = "_lock_id" LOCK_TIME_KEY = "_lock_time" @@ -98,7 +97,7 @@ def acquire(self): msg = ( f"The lock was broken. Previous lock id: {self.get_lock_id(result)}" ) - warnings.warn(msg) + warnings.warn(msg, stacklevel=2) self.locked_document = result @@ -119,7 +118,7 @@ def release(self, exc_type, exc_val, exc_tb): # Check if the lock was successfully released if result.modified_count == 0: msg = f"Could not release lock for document {self.locked_document['_id']}" - warnings.warn(msg) + warnings.warn(msg, stacklevel=2) self.locked_document = None @@ -128,6 +127,5 @@ def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): - if self.locked_document: self.release(exc_type, exc_val, exc_tb) diff --git a/src/jobflow_remote/utils/log.py b/src/jobflow_remote/utils/log.py index 9dec2ec3..81cd93f8 100644 --- a/src/jobflow_remote/utils/log.py +++ b/src/jobflow_remote/utils/log.py @@ -1,5 +1,7 @@ """Tools for logging.""" +from __future__ import annotations + import logging import logging.config from pathlib import Path diff --git a/tests/test_jobflow_remote.py b/tests/test_jobflow_remote.py index fce15f0a..64ff8957 100644 --- a/tests/test_jobflow_remote.py +++ b/tests/test_jobflow_remote.py @@ -2,4 +2,14 @@ def test_version(): - assert __version__ == "0.1.0" + assert __version__ == "0.0.1" + + +def test_imports(): + """This test triggers all the top-level imports by importing + the global `SETTINGS`. + + """ + from jobflow_remote import SETTINGS # noqa + + ...