feat: Hatch build hook for Polylith

bases/polylith/cli/create.py

@@ -1,5 +1,4 @@
 from pathlib import Path
-from typing import Union
 
 from polylith import project, repo
 from polylith.bricks import base, component
@@ -29,8 +28,10 @@ def component_command(
     create(name, description, component.create_component)
 
 
-def _create_project(root: Path, _ns: str, name: str, description: Union[str, None]):
+def _create_project(root: Path, options: dict):
     root_pyproject: dict = project.get_toml(root / repo.default_toml)
+    name = options["package"]
+    description = options["description"]
 
     if repo.is_poetry(root_pyproject):
         template = project.templates.poetry_pyproject

bases/polylith/hatch_hooks/hooks.py

@@ -0,0 +1,8 @@
+from hatchling.plugin import hookimpl
+
+from polylith.hatch.hooks.bricks import PolylithBricksHook
+
+
+@hookimpl
+def hatch_register_build_hook():
+    return PolylithBricksHook

components/polylith/hatch/__init__.py

@@ -0,0 +1,3 @@
+from polylith.hatch import hooks
+
+__all__ = ["hooks"]

components/polylith/hatch/core.py

@@ -0,0 +1,41 @@
+from pathlib import Path
+from typing import List, Union
+
+from polylith import parsing
+
+
+def get_work_dir(config: dict) -> Path:
+    work_dir = config.get("work-dir", ".polylith_tmp")
+
+    return Path(work_dir)
+
+
+def parse_namespace(bricks: dict) -> str:
+    namespaces = parsing.parse_brick_namespace_from_path(bricks)
+
+    return next(namespace for namespace in namespaces)
+
+
+def copy_brick(source: str, brick: str, tmp_dir: Path) -> Path:
+    destination = Path(tmp_dir / brick).as_posix()
+
+    return parsing.copy_brick(source, destination)
+
+
+def rewrite_module(module: Path, ns: str, top_ns: str) -> Union[str, None]:
+    was_rewritten = parsing.rewrite_module(module, ns, top_ns)
+
+    return f"{module.parent.name}/{module.name}" if was_rewritten else None
+
+
+def rewrite_modules(path: Path, ns: str, top_ns: str) -> List[str]:
+    """Rewrite modules in bricks with new top namespace
+
+    returns a list of bricks that was rewritten
+    """
+
+    modules = path.glob("**/*.py")
+
+    res = [rewrite_module(module, ns, top_ns) for module in modules]
+
+    return [r for r in res if r]

components/polylith/hatch/hooks/bricks.py

@@ -0,0 +1,49 @@
+import shutil
+from pathlib import Path
+from typing import Any, Dict
+
+from hatchling.builders.hooks.plugin.interface import BuildHookInterface
+from polylith import repo, toml
+from polylith.hatch import core
+
+
+class PolylithBricksHook(BuildHookInterface):
+    PLUGIN_NAME = "polylith-bricks"
+
+    def initialize(self, _version: str, build_data: Dict[str, Any]) -> None:
+        top_ns = self.config.get("top-namespace")
+        work_dir = core.get_work_dir(self.config)
+        pyproject = Path(f"{self.root}/{repo.default_toml}")
+
+        print(f"Using {pyproject.as_posix()}.")
+
+        data = toml.read_toml_document(pyproject)
+        bricks = toml.get_project_packages_from_polylith_section(data)
+
+        if not bricks:
+            print("No bricks found.")
+            return
+
+        if not top_ns:
+            build_data["force_include"] = bricks
+            return
+
+        ns = core.parse_namespace(bricks)
+
+        for source, brick in bricks.items():
+            path = core.copy_brick(source, brick, work_dir)
+            rewritten_bricks = core.rewrite_modules(path, ns, top_ns)
+
+            for item in rewritten_bricks:
+                print(f"Updated {item} with new top namespace for local imports.")
+
+        key = work_dir.as_posix()
+        build_data["force_include"][key] = top_ns
+
+    def finalize(self, *args, **kwargs) -> None:
+        work_dir = core.get_work_dir(self.config)
+
+        if not work_dir.exists() or not work_dir.is_dir():
+            return
+
+        shutil.rmtree(work_dir.as_posix())

components/polylith/parsing/__init__.py

@@ -0,0 +1,4 @@
+from polylith.parsing.core import copy_brick, parse_brick_namespace_from_path
+from polylith.parsing.rewrite import rewrite_module
+
+__all__ = ["copy_brick", "parse_brick_namespace_from_path", "rewrite_module"]

components/polylith/parsing/core.py

@@ -0,0 +1,24 @@
+import shutil
+from pathlib import Path
+
+
+def copy_brick(source: str, destination: str) -> Path:
+    ignore = shutil.ignore_patterns(
+        "*.pyc",
+        "__pycache__",
+        ".venv",
+        ".mypy_cache",
+        ".pytest_cache",
+        "node_modules",
+        ".git",
+    )
+
+    res = shutil.copytree(source, destination, ignore=ignore, dirs_exist_ok=True)
+
+    return Path(res)
+
+
+def parse_brick_namespace_from_path(bricks: dict) -> set:
+    parts = {str.split(v, "/")[0] for v in bricks.values()}
+
+    return parts

components/polylith/parsing/rewrite.py

@@ -0,0 +1,60 @@
+import ast
+from pathlib import Path
+
+
+def create_namespace_path(top_ns: str, current: str) -> str:
+    top_ns_module_path = top_ns.replace("/", ".")
+    return f"{top_ns_module_path}.{current}"
+
+
+def mutate_import(node: ast.Import, ns: str, top_ns: str) -> bool:
+    did_mutate = False
+
+    for alias in node.names:
+        if alias.name == ns:
+            alias.name = create_namespace_path(top_ns, alias.name)
+            did_mutate = True
+
+    return did_mutate
+
+
+def mutate_import_from(node: ast.ImportFrom, ns: str, top_ns: str) -> bool:
+    did_mutate = False
+
+    if not node.module or node.level != 0:
+        return did_mutate
+
+    if node.module == ns or node.module.startswith(f"{ns}."):
+        node.module = create_namespace_path(top_ns, node.module)
+        did_mutate = True
+
+    return did_mutate
+
+
+def mutate_imports(node: ast.AST, ns: str, top_ns: str) -> bool:
+    if isinstance(node, ast.Import):
+        return mutate_import(node, ns, top_ns)
+
+    if isinstance(node, ast.ImportFrom):
+        return mutate_import_from(node, ns, top_ns)
+
+    return False
+
+
+def rewrite_module(source: Path, ns: str, top_ns: str) -> bool:
+    file_path = source.as_posix()
+
+    with open(file_path, "r", encoding="utf-8") as f:
+        tree = ast.parse(f.read(), source.name)
+
+    res = {mutate_imports(node, ns, top_ns) for node in ast.walk(tree)}
+
+    if True in res:
+        rewritten_source_code = ast.unparse(tree)  # type: ignore[attr-defined]
+
+        with open(file_path, "w", encoding="utf-8", newline="") as f:
+            f.write(rewritten_source_code)
+
+        return True
+
+    return False

components/polylith/project/__init__.py

@@ -1,17 +1,11 @@
 from polylith.project import templates
 from polylith.project.create import create_project
-from polylith.project.get import (
-    get_packages_for_projects,
-    get_project_dependencies,
-    get_project_name,
-    get_toml,
-)
+from polylith.project.get import get_packages_for_projects, get_project_name, get_toml
 from polylith.project.parser import parse_package_paths
 
 __all__ = [
     "create_project",
     "get_packages_for_projects",
-    "get_project_dependencies",
     "get_project_name",
     "get_toml",
     "parse_package_paths",

components/polylith/project/get.py

@@ -1,57 +1,9 @@
-import re
 from functools import lru_cache
 from pathlib import Path
-from typing import Any, List
+from typing import List
 
 import tomlkit
-from polylith import repo, workspace
-
-
-def transform_to_package(namespace: str, include: str) -> dict:
-    path, _separator, brick = str.partition(include, f"/{namespace}/")
-
-    return {"include": f"{namespace}/{brick}", "from": path}
-
-
-def find_by_key(data: dict, key: str) -> Any:
-    if key in data.keys():
-        return data[key]
-
-    filtered = {k: v for k, v in data.items() if isinstance(data[k], dict)}
-
-    res = (find_by_key(data[k], key) for k in filtered.keys())
-
-    return next((r for r in res if r), None)
-
-
-def get_project_packages_from_polylith_section(data) -> dict:
-    bricks = data["tool"].get("polylith", {}).get("bricks")
-
-    return bricks if isinstance(bricks, dict) else {}
-
-
-def get_hatch_project_packages(data) -> dict:
-    hatch_data = data["tool"]["hatch"]
-    build_data = hatch_data.get("build", {}) if isinstance(hatch_data, dict) else {}
-
-    force_included = build_data.get("force-include")
-
-    if force_included:
-        return force_included
-
-    return get_project_packages_from_polylith_section(data)
-
-
-def get_project_package_includes(namespace: str, data) -> List[dict]:
-    if repo.is_poetry(data):
-        return data["tool"]["poetry"].get("packages", [])
-
-    if repo.is_hatch(data):
-        includes = get_hatch_project_packages(data)
-
-        return [transform_to_package(namespace, key) for key in includes.keys()]
-
-    return []
+from polylith import repo, toml, workspace
 
 
 def get_project_name(data) -> str:
@@ -61,23 +13,9 @@ def get_project_name(data) -> str:
     return data["tool"]["poetry"]["name"]
 
 
-def get_project_dependencies(data) -> dict:
-    if repo.is_poetry(data):
-        deps = data["tool"]["poetry"].get("dependencies", [])
-
-        items = set(deps.keys())
-    else:
-        deps = data["project"].get("dependencies", [])
-
-        items = {re.split(r"[\^~=!<>]", dep)[0] for dep in deps}
-
-    return {"items": items, "source": repo.default_toml}
-
-
 @lru_cache
 def get_toml(path: Path) -> tomlkit.TOMLDocument:
-    with path.open(encoding="utf-8", errors="ignore") as f:
-        return tomlkit.loads(f.read())
+    return toml.read_toml_document(path)
 
 
 def get_project_files(root: Path) -> dict:
@@ -104,16 +42,16 @@ def get_toml_files(root: Path) -> List[dict]:
 
 
 def get_packages_for_projects(root: Path) -> List[dict]:
-    tomls = get_toml_files(root)
+    toml_files = get_toml_files(root)
     namespace = workspace.parser.get_namespace_from_config(root)
 
     return [
         {
             "name": get_project_name(d["toml"]),
-            "packages": get_project_package_includes(namespace, d["toml"]),
+            "packages": toml.get_project_package_includes(namespace, d["toml"]),
             "path": d["path"],
             "type": d["type"],
-            "deps": get_project_dependencies(d["toml"]),
+            "deps": toml.get_project_dependencies(d["toml"]),
         }
-        for d in tomls
+        for d in toml_files
     ]

components/polylith/repo/repo.py

@@ -57,12 +57,18 @@ def get_workspace_root(cwd: Path) -> Path:
     return root
 
 
+def has_build_requires(pyproject: dict, value: str) -> bool:
+    backend = pyproject.get("build-system", {}).get("build-backend", {})
+
+    return value in backend
+
+
 def is_poetry(pyproject: dict) -> bool:
-    return pyproject.get("tool", {}).get("poetry") is not None
+    return has_build_requires(pyproject, "poetry")
 
 
 def is_hatch(pyproject: dict) -> bool:
-    return pyproject.get("tool", {}).get("hatch") is not None
+    return has_build_requires(pyproject, "hatchling")
 
 
 def is_pep_621_ready(pyproject: dict) -> bool:

components/polylith/toml/__init__.py

@@ -0,0 +1,13 @@
+from polylith.toml.core import (
+    get_project_dependencies,
+    get_project_package_includes,
+    get_project_packages_from_polylith_section,
+    read_toml_document,
+)
+
+__all__ = [
+    "get_project_dependencies",
+    "get_project_package_includes",
+    "get_project_packages_from_polylith_section",
+    "read_toml_document",
+]