Further improvements

Author: Martin-Molinero

lean/click.py

@@ -11,19 +11,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from datetime import datetime
 from pathlib import Path
 from typing import Optional, List
 
-import click
+from click import Command, Context, Parameter, ParamType, Option as ClickOption
 
 from lean.constants import DEFAULT_LEAN_CONFIG_FILE_NAME
 from lean.container import container
 from lean.models.errors import MoreInfoError
 from lean.models.logger import Option
 
-
-class LeanCommand(click.Command):
+class LeanCommand(Command):
     """A click.Command wrapper with some Lean CLI customization."""
 
     def __init__(self,
@@ -54,17 +52,21 @@ def __init__(self,
         self.context_settings["ignore_unknown_options"] = allow_unknown_options
         self.context_settings["allow_extra_args"] = allow_unknown_options
 
-    def invoke(self, ctx: click.Context):
+    def invoke(self, ctx: Context):
         if self._requires_lean_config:
-            lean_config_manager = container.lean_config_manager()
+
+            from time import time
+            start = time()
+
+            lean_config_manager = container.lean_config_manager
             try:
                 # This method will raise an error if the directory cannot be found
                 lean_config_manager.get_cli_root_directory()
             except Exception:
                 # Use one of the cached Lean config locations to avoid having to abort the command
                 lean_config_paths = lean_config_manager.get_known_lean_config_paths()
                 if len(lean_config_paths) > 0:
-                    lean_config_path = container.logger().prompt_list("Select the Lean configuration file to use", [
+                    lean_config_path = container.logger.prompt_list("Select the Lean configuration file to use", [
                         Option(id=p, label=str(p)) for p in lean_config_paths
                     ])
                     lean_config_manager.set_default_lean_config_path(lean_config_path)
@@ -75,37 +77,36 @@ def invoke(self, ctx: click.Context):
                         "https://www.lean.io/docs/v2/lean-cli/key-concepts/troubleshooting#02-Common-Errors"
                     )
 
-        import sys
-        if self._requires_docker and "pytest" not in sys.modules:
-            import os
-            is_system_linux = container.platform_manager().is_system_linux()
-
-            # The CLI uses temporary directories in /tmp because sometimes it may leave behind files owned by root
-            # These files cannot be deleted by the CLI itself, so we rely on the OS to empty /tmp on reboot
-            # The Snap version of Docker does not provide access to files outside $HOME, so we can't support it
-            if is_system_linux:
-                import shutil
-                docker_path = shutil.which("docker")
+        if self._requires_docker:
+            from sys import modules, executable, argv
+            if "pytest" not in modules and container.platform_manager.is_system_linux():
+                from shutil import which
+                from os import getuid, execlp
+                # The CLI uses temporary directories in /tmp because sometimes it may leave behind files owned by root
+                # These files cannot be deleted by the CLI itself, so we rely on the OS to empty /tmp on reboot
+                # The Snap version of Docker does not provide access to files outside $HOME, so we can't support it
+
+                docker_path = which("docker")
                 if docker_path is not None and docker_path.startswith("/snap"):
                     raise MoreInfoError(
                         "The Lean CLI does not work with the Snap version of Docker, please re-install Docker via the official installation instructions",
                         "https://docs.docker.com/engine/install/")
 
-            # A usual Docker installation on Linux requires the user to use sudo to run Docker
-            # If we detect that this is the case and the CLI was started without sudo we elevate automatically
-            if is_system_linux and os.getuid() != 0 and container.docker_manager().is_missing_permission():
-                container.logger().info(
-                    "This command requires access to Docker, you may be asked to enter your password")
+                # A usual Docker installation on Linux requires the user to use sudo to run Docker
+                # If we detect that this is the case and the CLI was started without sudo we elevate automatically
+                if getuid() != 0 and container.docker_manager.is_missing_permission():
+                    container.logger.info(
+                        "This command requires access to Docker, you may be asked to enter your password")
 
-                args = ["sudo", "--preserve-env=HOME", sys.executable, *sys.argv]
-                os.execlp(args[0], *args)
+                    args = ["sudo", "--preserve-env=HOME", executable, *argv]
+                    execlp(args[0], *args)
 
         if self._allow_unknown_options:
-            import itertools
+            from itertools import chain
             # Unknown options are passed to ctx.args and need to be parsed manually
             # We parse them to ctx.params so they're available like normal options
             # Because of this all commands with allow_unknown_options=True must have a **kwargs argument
-            arguments = list(itertools.chain(*[arg.split("=") for arg in ctx.args]))
+            arguments = list(chain(*[arg.split("=") for arg in ctx.args]))
 
             skip_next = False
             for index in range(len(arguments) - 1):
@@ -119,7 +120,7 @@ def invoke(self, ctx: click.Context):
                     ctx.params[option] = value
                     skip_next = True
 
-        update_manager = container.update_manager()
+        update_manager = container.update_manager
         update_manager.show_announcements()
 
         result = super().invoke(ctx)
@@ -128,20 +129,20 @@ def invoke(self, ctx: click.Context):
 
         return result
 
-    def get_params(self, ctx: click.Context):
+    def get_params(self, ctx: Context):
         params = super().get_params(ctx)
 
         # Add --lean-config option if the command requires a Lean config
         if self._requires_lean_config:
-            params.insert(len(params) - 1, click.Option(["--lean-config"],
+            params.insert(len(params) - 1, ClickOption(["--lean-config"],
                                                         type=PathParameter(exists=True, file_okay=True, dir_okay=False),
                                                         help=f"The Lean configuration file that should be used (defaults to the nearest {DEFAULT_LEAN_CONFIG_FILE_NAME})",
                                                         expose_value=False,
                                                         is_eager=True,
                                                         callback=self._parse_config_option))
 
         # Add --verbose option
-        params.insert(len(params) - 1, click.Option(["--verbose"],
+        params.insert(len(params) - 1, ClickOption(["--verbose"],
                                                     help="Enable debug logging",
                                                     is_flag=True,
                                                     default=False,
@@ -151,20 +152,20 @@ def get_params(self, ctx: click.Context):
 
         return params
 
-    def _parse_config_option(self, ctx: click.Context, param: click.Parameter, value: Optional[Path]) -> None:
+    def _parse_config_option(self, ctx: Context, param: Parameter, value: Optional[Path]) -> None:
         """Parses the --config option."""
         if value is not None:
-            lean_config_manager = container.lean_config_manager()
+            lean_config_manager = container.lean_config_manager
             lean_config_manager.set_default_lean_config_path(value)
 
-    def _parse_verbose_option(self, ctx: click.Context, param: click.Parameter, value: Optional[bool]) -> None:
+    def _parse_verbose_option(self, ctx: Context, param: Parameter, value: Optional[bool]) -> None:
         """Parses the --verbose option."""
         if value:
-            logger = container.logger()
+            logger = container.logger
             logger.debug_logging_enabled = True
 
 
-class PathParameter(click.ParamType):
+class PathParameter(ParamType):
     """A limited version of click.Path which uses pathlib.Path."""
 
     def __init__(self, exists: bool = False, file_okay: bool = True, dir_okay: bool = True):
@@ -188,10 +189,10 @@ def __init__(self, exists: bool = False, file_okay: bool = True, dir_okay: bool
             self.name = "path"
             self._path_type = "Path"
 
-    def convert(self, value: str, param: click.Parameter, ctx: click.Context) -> Path:
+    def convert(self, value: str, param: Parameter, ctx: Context) -> Path:
         path = Path(value).expanduser().resolve()
 
-        if not container.path_manager().is_path_valid(path):
+        if not container.path_manager.is_path_valid(path):
             self.fail(f"{self._path_type} '{value}' is not a valid path.", param, ctx)
 
         if self._exists and not path.exists():
@@ -206,15 +207,16 @@ def convert(self, value: str, param: click.Parameter, ctx: click.Context) -> Pat
         return path
 
 
-class DateParameter(click.ParamType):
+class DateParameter(ParamType):
     """A click parameter which returns datetime.datetime objects and requires yyyyMMdd input."""
 
     name = "date"
 
-    def get_metavar(self, param: click.Parameter) -> str:
+    def get_metavar(self, param: Parameter) -> str:
         return "[yyyyMMdd]"
 
-    def convert(self, value: str, param: click.Parameter, ctx: click.Context) -> datetime:
+    def convert(self, value: str, param: Parameter, ctx: Context):
+        from datetime import datetime
         for date_format in ["%Y%m%d", "%Y-%m-%d"]:
             try:
                 return datetime.strptime(value, date_format)
@@ -229,7 +231,9 @@ def ensure_options(options: List[str]) -> None:
 
     :param options: the Python names of the options that must have values
     """
-    ctx = click.get_current_context()
+    from click import get_current_context
+
+    ctx = get_current_context()
 
     missing_options = []
     for key, value in ctx.params.items():
@@ -251,7 +255,9 @@ def ensure_options(options: List[str]) -> None:
         option = next(param for param in ctx.command.params if param.name == name)
         help_records.append(option.get_help_record(ctx))
 
-    help_formatter = click.HelpFormatter(max_width=120)
+    from click import HelpFormatter
+
+    help_formatter = HelpFormatter(max_width=120)
     help_formatter.write_dl(help_records)
 
     raise RuntimeError(f"""

lean/commands/backtest.py

@@ -16,6 +16,7 @@
 from pathlib import Path
 from typing import Optional
 from click import command, option, argument, Choice
+
 from lean.click import LeanCommand, PathParameter
 from lean.constants import DEFAULT_ENGINE_IMAGE, LEAN_ROOT_PATH
 from lean.container import container, Logger
@@ -41,7 +42,7 @@ def _migrate_python_pycharm(logger: Logger, project_dir: Path) -> None:
     if not workspace_xml_path.is_file():
         return
 
-    xml_manager = container.xml_manager()
+    xml_manager = container.xml_manager
     current_content = xml_manager.parse(workspace_xml_path.read_text(encoding="utf-8"))
 
     config = current_content.find('.//configuration[@name="Debug with Lean CLI"]')
@@ -55,7 +56,7 @@ def _migrate_python_pycharm(logger: Logger, project_dir: Path) -> None:
     made_changes = False
     has_library_mapping = False
 
-    library_dir = container.lean_config_manager().get_cli_root_directory() / "Library"
+    library_dir = container.lean_config_manager.get_cli_root_directory() / "Library"
 
     if library_dir.is_dir():
         library_dir = f"$PROJECT_DIR$/{path.relpath(library_dir, project_dir)}".replace("\\", "/")
@@ -82,7 +83,7 @@ def _migrate_python_pycharm(logger: Logger, project_dir: Path) -> None:
     if made_changes:
         workspace_xml_path.write_text(xml_manager.to_string(current_content), encoding="utf-8")
 
-        logger = container.logger()
+        logger = container.logger
         logger.warn("Your run configuration has been updated to work with the latest version of LEAN")
         logger.warn("Please restart the debugger in PyCharm and run this command again")
 
@@ -105,7 +106,7 @@ def _migrate_python_vscode(project_dir: Path) -> None:
     made_changes = False
     has_library_mapping = False
 
-    library_dir = container.lean_config_manager().get_cli_root_directory() / "Library"
+    library_dir = container.lean_config_manager.get_cli_root_directory() / "Library"
     if not library_dir.is_dir():
         library_dir = None
 
@@ -132,7 +133,7 @@ def _migrate_csharp_rider(logger: Logger, project_dir: Path) -> None:
     from click import Abort
 
     made_changes = False
-    xml_manager = container.xml_manager()
+    xml_manager = container.xml_manager
 
     for dir_name in [f".idea.{project_dir.stem}", f".idea.{project_dir.stem}.dir"]:
         workspace_xml_path = project_dir / ".idea" / dir_name / ".idea" / "workspace.xml"
@@ -155,7 +156,7 @@ def _migrate_csharp_rider(logger: Logger, project_dir: Path) -> None:
         made_changes = True
 
     if made_changes:
-        container.project_manager().generate_rider_config()
+        container.project_manager.generate_rider_config()
 
         logger.warn("Your run configuration has been updated to work with the .NET 5 version of LEAN")
         logger.warn("Please restart Rider and start debugging again")
@@ -207,7 +208,7 @@ def _migrate_csharp_csproj(project_dir: Path) -> None:
     if csproj_path is None:
         return
 
-    xml_manager = container.xml_manager()
+    xml_manager = container.xml_manager
 
     current_content = xml_manager.parse(csproj_path.read_text(encoding="utf-8"))
     if current_content.find(".//PropertyGroup/DefaultItemExcludes") is not None:
@@ -230,12 +231,12 @@ def _select_organization() -> QCMinimalOrganization:
 
     :return: the selected organization
     """
-    api_client = container.api_client()
+    api_client = container.api_client
 
     organizations = api_client.organizations.get_all()
     options = [Option(id=organization, label=organization.name) for organization in organizations]
 
-    logger = container.logger()
+    logger = container.logger
     return logger.prompt_list("Select the organization to purchase and download data with", options)
 
 
@@ -304,10 +305,10 @@ def backtest(project: Path,
     You can override this using the --image option.
     Alternatively you can set the default engine image for all commands using `lean config set engine-image <image>`.
     """
-    logger = container.logger()
-    project_manager = container.project_manager()
+    logger = container.logger
+    project_manager = container.project_manager
     algorithm_file = project_manager.find_algorithm_file(Path(project))
-    lean_config_manager = container.lean_config_manager()
+    lean_config_manager = container.lean_config_manager
     from datetime import datetime
     if output is None:
         output = algorithm_file.parent / "backtests" / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
@@ -343,21 +344,21 @@ def backtest(project: Path,
 
     lean_config_manager.configure_data_purchase_limit(lean_config, data_purchase_limit)
 
-    cli_config_manager = container.cli_config_manager()
-    project_config_manager = container.project_config_manager()
+    cli_config_manager = container.cli_config_manager
+    project_config_manager = container.project_config_manager
 
     project_config = project_config_manager.get_project_config(algorithm_file.parent)
     engine_image = cli_config_manager.get_engine_image(image or project_config.get("engine-image", None))
 
     if str(engine_image) != DEFAULT_ENGINE_IMAGE:
         logger.warn(f'A custom engine image: "{engine_image}" is being used!')
 
-    container.update_manager().pull_docker_image_if_necessary(engine_image, update)
+    container.update_manager.pull_docker_image_if_necessary(engine_image, update)
 
     if not output.exists():
         output.mkdir(parents=True)
 
-    output_config_manager = container.output_config_manager()
+    output_config_manager = container.output_config_manager
     lean_config["algorithm-id"] = str(output_config_manager.get_backtest_id(output))
 
     # Set backtest name
@@ -367,7 +368,7 @@ def backtest(project: Path,
     if python_venv is not None and python_venv != "":
         lean_config["python-venv"] = f'{"/" if python_venv[0] != "/" else ""}{python_venv}'
 
-    lean_runner = container.lean_runner()
+    lean_runner = container.lean_runner
     lean_runner.run_lean(lean_config,
                          "backtesting",
                          algorithm_file,