lint with black

Author: ninthreezy

datatorch/agent/__init__.py

@@ -54,7 +54,7 @@ def setup_logging() -> None:
 
 
 async def _exit_jobs() -> None:
-    """ Exits active running agent jobs """
+    """Exits active running agent jobs"""
     logger.info(f"Exiting {len(tasks)} active jobs.")
 
     for task in tasks:
@@ -77,7 +77,7 @@ async def _close_transport(transport: WebsocketsTransport):
 
 
 async def _exit_tasks() -> None:
-    """ Exits all active asyncio tasks """
+    """Exits all active asyncio tasks"""
     current_task = asyncio.Task.current_task()
     all_tasks = asyncio.Task.all_tasks()
     not_current_tasks = [task for task in all_tasks if task is not current_task]
@@ -87,7 +87,7 @@ async def _exit_tasks() -> None:
 
 
 async def start() -> None:
-    """ Creates and runs an agent. """
+    """Creates and runs an agent."""
     setup_logging()
 
     click.echo(
@@ -131,7 +131,7 @@ async def start() -> None:
 
 
 async def stop() -> None:
-    """ Stop all run tasks. """
+    """Stop all run tasks."""
 
     print(" ")
 

datatorch/agent/agent.py

@@ -46,7 +46,7 @@ def _init_threads(self):
         tasks.append(task)
 
     async def process_loop(self):
-        """ Waits for jobs from server. """
+        """Waits for jobs from server."""
         logger.info("Waiting for jobs.")
         async for job_request in self.api.agent_jobs():
             loop = asyncio.get_event_loop()
@@ -55,7 +55,7 @@ async def process_loop(self):
             tasks.append(task)
 
     async def _run_job(self, job: AgentJobConfig):
-        """ Runs a job """
+        """Runs a job"""
         job_id = job.get("id")
         job_name = job.get("name")
         job_steps = job.get("steps")

datatorch/agent/client.py

@@ -58,7 +58,7 @@ def __init__(self, session: AsyncClientSession):
         self.session = session
 
     def agent_jobs(self):
-        """ Subscriptions to the agent job assignment namespace. """
+        """Subscriptions to the agent job assignment namespace."""
         # fmt: off
         sub = gql("""
             subscription {
@@ -220,7 +220,7 @@ def upload_step_logs(self, step_id: str, logs: List[Log]):
         return self.execute(mutate, params={"id": step_id, "logs": logs})
 
     async def execute(self, query, *args, params: dict = {}, **kwargs) -> dict:
-        """ Wrapper around execute """
+        """Wrapper around execute"""
         removed_none = dict((k, v) for k, v in params.items() if v is not None)
         if type(query) == str:
             query = gql(query)

datatorch/agent/directory.py

@@ -7,7 +7,7 @@
 class AgentDirectory(object):
     @staticmethod
     def path() -> str:
-        """ Returns the agents directory """
+        """Returns the agents directory"""
         path = folder.get_app_dir()
         return os.getenv("DATATORCH_AGENT_PATH", os.path.join(path, "agent"))
 
@@ -36,12 +36,12 @@ def runs_dir(self):
 
     @property
     def logs_dir(self):
-        """ Directory where agent logs are stored. """
+        """Directory where agent logs are stored."""
         return os.path.join(self.dir, "logs")
 
     @property
     def db_dir(self):
-        """ Sqlite database are stored. """
+        """Sqlite database are stored."""
         return os.path.join(self.dir, "db")
 
     @property
@@ -55,7 +55,7 @@ def projects_dir(self):
 
     @property
     def actions_dir(self):
-        """ Directory where actions are stored. """
+        """Directory where actions are stored."""
         return os.path.join(self.dir, "actions")
 
     def open(self, file: str, mode: str):
@@ -65,13 +65,13 @@ def action_dir(self, name: str, version: str):
         return os.path.join(self.actions_dir, *name.lower().split("/"), version)
 
     def run_dir(self, task_id: str):
-        """ Returns the directory for a given task """
+        """Returns the directory for a given task"""
         path = os.path.join(self.runs_dir, task_id)
         mkdir_exists(path)
         return path
 
     def project_dir(self, project_id: str):
-        """ Returns the directory for a given project """
+        """Returns the directory for a given project"""
         path = os.path.join(self.projects_dir, project_id)
         mkdir_exists(path)
         return path

datatorch/agent/monitoring.py

@@ -21,7 +21,7 @@ def print_stats(metrics):
 class AgentSystemStats(object):
     @staticmethod
     def initial_stats():
-        """ Returns stats that do not change over time. """
+        """Returns stats that do not change over time."""
         # initialize averaging
         psutil.cpu_percent()
         cpu_freq = psutil.cpu_freq()

datatorch/agent/pipelines/job/job.py

@@ -35,7 +35,7 @@ async def update(self, status: str) -> None:
         await self.agent.api.update_job(variables)
 
     async def run(self, variables: Variables):
-        """ Runs each step of the job. """
+        """Runs each step of the job."""
         steps = Step.from_dict_list(self.config.get("steps", []), job=self)
         await self.update("RUNNING")
 

datatorch/agent/pipelines/pipeline.py

@@ -22,7 +22,7 @@ def from_yaml(cls, path, agent: "Agent" = None):
 
     @classmethod
     def from_config(cls, config: Union[str, dict], agent: "Agent" = None):
-        """ Creates a pipeline from a config file. """
+        """Creates a pipeline from a config file."""
         if isinstance(config, str):
             cf = yaml.load(config, Loader=yaml.FullLoader)
         else:
@@ -35,6 +35,6 @@ def __init__(self, config: dict, agent: "Agent" = None):
         self.agent = agent
 
     async def run(self, job_config: dict):
-        """ Runs a job. """
+        """Runs a job."""
         # await Job(job_config, agent=self.agent).run()
         pass

datatorch/agent/pipelines/runner/factory.py

@@ -23,7 +23,7 @@ class RunnerCreateError(Exception):
 class RunnerFactory(object):
     @staticmethod
     def create(action, config: dict) -> Runner:
-        """ Makes runners to 'use' strings found in config.yaml files. """
+        """Makes runners to 'use' strings found in config.yaml files."""
         use = config.get("using")
         if use is None:
             raise RunnerCreateError("Action 'use' property not specified.")

datatorch/agent/pipelines/runner/runner.py

@@ -30,11 +30,11 @@ async def execute(self) -> Awaitable[dict]:
         raise NotImplementedError("This method must be implemented.")
 
     def action_dir(self):
-        """ Changes the current work directory to the actions directory. """
+        """Changes the current work directory to the actions directory."""
         os.chdir(self.action.dir)
 
     async def monitor_cmd(self, command: str):
-        """ Excutes a command and monitors stdout for variables and logging. """
+        """Excutes a command and monitors stdout for variables and logging."""
         process = await self.run_cmd(command)
 
         async for log in process.stdout:
@@ -48,11 +48,11 @@ async def monitor_cmd(self, command: str):
             )
 
     def get(self, key: str, default=None):
-        """ Gets a string from config and renders template. """
+        """Gets a string from config and renders template."""
         return self.variables.render(self.config.get(key, default))
 
     def check_for_output(self, string: str) -> bool:
-        """ Parse output variable from string if valid. """
+        """Parse output variable from string if valid."""
         # ::varname::value tranlatest to varname = value
         result = string.split("::", 2)
         if len(result) != 3:
@@ -62,7 +62,7 @@ def check_for_output(self, string: str) -> bool:
         return True
 
     async def run_cmd(self, command: str, wait: bool = True):
-        """ Runs a command using asyncio """
+        """Runs a command using asyncio"""
         process = await asyncio.create_subprocess_shell(
             command,
             shell=True,

datatorch/agent/pipelines/step/step.py

@@ -109,17 +109,17 @@ async def log_uploader(self):
 
     @property
     def api(self):
-        """ Agent API Client if it exists. """
+        """Agent API Client if it exists."""
         return self.job and self.job.api
 
     def log(self, message: str):
-        """ Records a log message. """
+        """Records a log message."""
         iso_date = datetime.now(timezone.utc).isoformat()[:-9] + "Z"
         self.logs.append(dict(createdAt=iso_date, message=message))  # type: ignore
         self.logger.info(message)
 
     async def upload_logs(self):
-        """ Uploads saved logs to webserver. """
+        """Uploads saved logs to webserver."""
         if self.id and len(self.logs) > 0:
             logs = self.logs
             self.logs = []