Fix lint issues across praktika

Author: maxknv

ci/tests/example_2/some_job_script.py

@@ -2,7 +2,7 @@
 from praktika.settings import Settings
 from praktika.utils import Shell, Utils
 
-# cache-bust: 2026-06-23
+# cache-bust: 2026-07-13
 
 if __name__ == "__main__":
     # 1. do some work

ci/tests/test_runner.py

@@ -269,6 +269,84 @@ def test_non_docker_job_does_not_mutate_parent_pythonpath(self):
         finally:
             os.environ.pop("PYTHONPATH", None)
 
+    def test_docker_job_mounts_installed_package_dir(self):
+        Path(_TEST_TEMP_DIR).mkdir(parents=True, exist_ok=True)
+
+        import sys
+        import types
+        from unittest import mock
+
+        with mock.patch.dict(sys.modules, {"requests": types.ModuleType("requests")}):
+            from praktika import runner
+
+            captured = {}
+
+            class DummyTeePopen:
+                def __init__(self, command, **kwargs):
+                    captured["command"] = command
+                    self.timeout_exceeded = False
+
+                def __enter__(self):
+                    return self
+
+                def __exit__(self, exc_type, exc_val, exc_tb):
+                    return False
+
+                def wait(self):
+                    return 0
+
+                def get_latest_log(self, max_lines=20):
+                    return ""
+
+            staged_package_dir = Path(runner._staged_praktika_package_dir())
+            staged_praktika_package = staged_package_dir / "praktika"
+            job = SimpleNamespace(
+                name="docker-job",
+                run_in_docker="example/image:latest",
+                timeout=1,
+                timeout_shell_cleanup=None,
+                enable_gh_auth=False,
+                command="echo ok",
+            )
+            workflow = SimpleNamespace(name="workflow")
+
+            with mock.patch.object(runner, "TeePopen", DummyTeePopen), mock.patch.object(
+                runner.Shell, "check", lambda *args, **kwargs: False
+            ), mock.patch.object(
+                runner.Shell, "run", lambda *args, **kwargs: None
+            ), mock.patch.object(
+                runner.Result,
+                "from_fs",
+                staticmethod(
+                    lambda *args, **kwargs: SimpleNamespace(
+                        is_completed=lambda: True,
+                        is_running=lambda: False,
+                        is_error=lambda: False,
+                        dump=lambda: None,
+                    )
+                ),
+            ), mock.patch.object(
+                runner._Environment,
+                "get",
+                staticmethod(
+                    lambda: SimpleNamespace(
+                        WORKFLOW_CONFIG=None,
+                        dump=lambda: None,
+                    )
+                ),
+            ):
+                rc = runner.Runner()._run(workflow=workflow, job=job, no_docker=False)
+
+            self.assertEqual(rc, 0)
+            self.assertTrue(staged_package_dir.is_dir())
+            self.assertTrue(staged_praktika_package.is_dir())
+            self.assertIn(
+                f"--volume {staged_package_dir}:{staged_package_dir}",
+                captured["command"],
+            )
+            self.assertIn(f"-e PYTHONPATH={staged_package_dir}", captured["command"])
+            self.assertNotIn("PYTHONPATH=.", captured["command"])
+
     def test_exit_code_result_synthesizes_fail_on_nonzero_exit(self):
         """enable_exit_code_result=True + script that exits non-zero
         without dumping a Result -> synthesized FAIL Result with the

ci/workflows/praktika_pr_advanced.py

@@ -8,7 +8,7 @@
 from ci.settings.settings import RunnerLabels
 from praktika.settings import Settings
 
-_HEAD_PRAKTIKA_VERSION = "0.1.8"
+_HEAD_PRAKTIKA_VERSION = "0.1.9"
 
 artifact = Artifact.Config(name="greet", type=Artifact.Type.S3, path="./artifact.txt")
 

praktika/__main__.py

@@ -359,7 +359,7 @@ def main(argv=None):
             for workflow in workflows:
                 print(
                     f"Workflow [{workflow.name}] has jobs:\n"
-                    "  \"" + f'"\n  "'.join([job.name for job in workflow.jobs]) + '"'
+                    '  "' + '"\n  "'.join([job.name for job in workflow.jobs]) + '"'
                     )
             Utils.exit_with_error("Job name is required to run a job.")
 

praktika/_environment.py

@@ -165,7 +165,7 @@ def from_env(cls) -> "_Environment":
                         if LINKED_PR_NUMBER
                         else ""
                     )
-                except:
+                except Exception:
                     LINKED_PR_NUMBER = 0
                     CHANGE_URL = ""
 
@@ -356,7 +356,7 @@ def get(cls):
                 env = cls.from_workflow_data()
                 env.dump()
                 return env
-            except FileNotFoundError as e:
+            except FileNotFoundError:
                 # For workflows without Config job
                 print(
                     f"NOTE: Workflow context file [{Settings.WORKFLOW_STATUS_FILE}] does not exist - read context from GH event"
@@ -428,7 +428,7 @@ def get_s3_prefix_static(cls, pr_number, branch, sha, workflow_name="", latest=F
             prefix = f"REFs/{branch}"
         assert sha or latest
         if latest:
-            prefix += f"/latest"
+            prefix += "/latest"
         elif sha:
             prefix += f"/{sha}"
         if cls._should_include_workflow_name_in_s3_prefix(workflow_name):

praktika/cache.py

@@ -53,7 +53,7 @@ def push_success_record(
         )
         assert (
             Settings.CACHE_S3_PATH
-        ), f"Setting CACHE_S3_PATH must be defined with enabled CI Cache"
+        ), "Setting CACHE_S3_PATH must be defined with enabled CI Cache"
         record_path = f"{Settings.CACHE_S3_PATH}/v{Settings.CACHE_VERSION}/{Utils.normalize_string(job_name)}/{job_digest}/{type_}"
         record_file = Path(Settings.TEMP_DIR) / type_
         record.dump(record_file)
@@ -69,7 +69,7 @@ def fetch_success(self, job_name, job_digest):
         type_ = Cache.CacheRecord.Type.SUCCESS
         assert (
             Settings.CACHE_S3_PATH
-        ), f"Setting CACHE_S3_PATH must be defined with enabled CI Cache"
+        ), "Setting CACHE_S3_PATH must be defined with enabled CI Cache"
         record_path = f"{Settings.CACHE_S3_PATH}/v{Settings.CACHE_VERSION}/{Utils.normalize_string(job_name)}/{job_digest}/{type_}"
         record_file_local_dir = (
             f"{Settings.CACHE_LOCAL_PATH}/{Utils.normalize_string(job_name)}/"

praktika/cidb.py

@@ -1,7 +1,6 @@
 import copy
 import dataclasses
 import json
-import os
 import time
 import urllib
 from typing import List, Optional
@@ -15,9 +14,10 @@
     if not Info().is_local_run:
         raise ex
     else:
-        print(
-            f"WARNING: 'requests' module is not installed: {ex}. CIDB will not work - ok for local runs only."
-        )
+            print(
+                "WARNING: 'requests' module is not installed: "
+                f"{ex}. CIDB will not work - ok for local runs only."
+            )
 
 from .result import Result
 from .settings import Settings
@@ -429,7 +429,7 @@ def check(self):
         # Create a session object
         params = {
             "database": Settings.CI_DB_DB_NAME,
-            "query": f"SELECT 1",
+            "query": "SELECT 1",
         }
         error = ""
         for retry in range(2):

praktika/gh.py

@@ -79,7 +79,7 @@ def get_changed_files(cls, strict=False) -> List[str]:
                 raise RuntimeError(
                     f"Failed to extract repository name from remote URL [{repo_url}]"
                 )
-            sha = Shell.get_output(f"git rev-parse HEAD", strict=True)
+            sha = Shell.get_output("git rev-parse HEAD", strict=True)
 
         assert repo_name
         print(repo_name)
@@ -797,7 +797,7 @@ def post_updateable_comment(
         else:
             if not only_update:
                 cmd = f"gh pr comment {pr} --body-file {temp_file_path}"
-                print(f"Create new comment")
+                print("Create new comment")
                 res = cls.do_command_with_retries(cmd)
             else:
                 print(
@@ -856,7 +856,7 @@ def get_pr_title_body_labels(cls, pr=None, repo=None):
             pr_data = json.loads(output)
             title = pr_data["title"]
             body = pr_data["body"]
-            labels = [l["name"] for l in pr_data["labels"]]
+            labels = [label["name"] for label in pr_data["labels"]]
         except Exception:
             print("ERROR: Failed to get PR data")
             traceback.print_exc()
@@ -1115,7 +1115,7 @@ class ResultSummaryForGH:
         sha: str = ""
         start_time: Optional[float] = None
         duration: Optional[float] = None
-        failed_results: List["ResultSummaryForGH"] = dataclasses.field(
+        failed_results: List["GH.ResultSummaryForGH"] = dataclasses.field(
             default_factory=list
         )
         info: str = ""

praktika/infrastructure/__init__.py

@@ -18,9 +18,14 @@
     "DedicatedHost",
     "EC2Instance",
     "IAMInstanceProfile",
+    "IAMRole",
     "ImageBuilder",
     "Lambda",
     "LaunchTemplate",
     "Components",
+    "ReportPage",
+    "SecretParameter",
     "SQSQueue",
+    "Storage",
+    "VPC",
 ]

praktika/infrastructure/autoscaling_group.py

@@ -1,6 +1,5 @@
 from ._utils import aws_client
-import json
-from dataclasses import asdict, dataclass, field
+from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional
 
 
@@ -56,8 +55,6 @@ def fetch(self):
             Raises:
                 Exception: If ASG does not exist or AWS API call fails
             """
-            import boto3
-
             asg_client = aws_client("autoscaling", self.region, self.name)
 
             resp = asg_client.describe_auto_scaling_groups(
@@ -125,8 +122,6 @@ def _resolve_subnet_ids(self) -> List[str]:
                     f"subnet_ids must be specified (non-empty) for ASG '{self.name}' or provide vpc_id/vpc_name for subnet discovery"
                 )
 
-            import boto3
-
             ec2 = aws_client("ec2", self.region, self.name)
 
             vpc_id = self.vpc_id
@@ -242,7 +237,6 @@ def deploy(self):
                 - This component intentionally does not try to manage every ASG attribute.
                 - It focuses on core runner-like ASG needs: subnets, LT, capacity, target groups, tags.
             """
-            import boto3
             from botocore.config import Config
 
             self.ext.pop("deferred_missing_launch_template", None)
@@ -392,7 +386,6 @@ def restart(self):
             return self
 
         def delete(self):
-            import boto3
             client = aws_client("autoscaling", self.region, self.name)
             try:
                 client.delete_auto_scaling_group(