Brick interfaces

components/polylith/check/__init__.py

@@ -1,3 +1,3 @@
-from polylith.check import collect, grouping, report
+from polylith.check import collect, report
 
-__all__ = ["collect", "grouping", "report"]
+__all__ = ["collect", "report"]

components/polylith/check/collect.py

@@ -1,13 +1,13 @@
 from pathlib import Path
 from typing import Set
 
-from polylith import check, imports, workspace
+from polylith import imports, workspace
 
 
 def extract_bricks(paths: Set[Path], ns: str) -> dict:
     all_imports = imports.fetch_all_imports(paths)
 
-    return check.grouping.extract_brick_imports(all_imports, ns)
+    return imports.extract_brick_imports(all_imports, ns)
 
 
 def with_unknown_components(root: Path, ns: str, brick_imports: dict) -> dict:

components/polylith/check/report.py

@@ -2,7 +2,7 @@
 from typing import Set
 
 from polylith import imports, libs, workspace
-from polylith.check import collect, grouping
+from polylith.check import collect
 from polylith.reporting import theme
 from rich.console import Console
 
@@ -78,8 +78,8 @@ def extract_collected_imports(
     ns: str, imports_in_bases: dict, imports_in_components: dict
 ) -> dict:
     brick_imports = {
-        "bases": grouping.extract_brick_imports(imports_in_bases, ns),
-        "components": grouping.extract_brick_imports(imports_in_components, ns),
+        "bases": imports.grouping.extract_brick_imports(imports_in_bases, ns),
+        "components": imports.grouping.extract_brick_imports(imports_in_components, ns),
     }
 
     third_party_imports = {

components/polylith/commands/deps.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 from typing import List, Set
 
-from polylith import bricks, deps, info
+from polylith import bricks, deps, info, interface
 
 
 def get_imports(root: Path, ns: str, bricks: dict) -> dict:
@@ -30,6 +30,17 @@ def get_components(root: Path, ns: str, project_data: dict) -> Set[str]:
     return pick_name(bricks.get_components_data(root, ns))
 
 
+def used_by_as_bricks(bricks: dict, brick_deps: dict) -> dict:
+    bases = bricks["bases"]
+    components = bricks["components"]
+
+    used_by = brick_deps["used_by"]
+    return {
+        "bases": {b for b in used_by if b in bases},
+        "components": {b for b in used_by if b in components},
+    }
+
+
 def run(root: Path, ns: str, options: dict):
     directory = options.get("directory")
     brick = options.get("brick")
@@ -53,13 +64,17 @@ def run(root: Path, ns: str, options: dict):
 
     if brick and imports.get(brick):
         brick_deps = bricks_deps[brick]
+        used_bricks = used_by_as_bricks(bricks, brick_deps)
+
         circular_deps = circular_bricks.get(brick)
 
         deps.print_brick_deps(brick, bricks, brick_deps, options)
 
         if circular_deps:
             deps.print_brick_with_circular_deps(brick, circular_deps, bricks)
 
+        interface.report.print_brick_interface_usage(root, ns, brick, used_bricks)
+
         return
 
     deps.print_deps(bricks, imports, options)

components/polylith/imports/__init__.py

@@ -1,13 +1,23 @@
+from polylith.imports.grouping import (
+    extract_brick_imports,
+    extract_brick_imports_with_namespaces,
+)
 from polylith.imports.parser import (
     extract_top_ns,
     fetch_all_imports,
+    fetch_api,
+    fetch_brick_import_usages,
     fetch_excluded_imports,
     list_imports,
 )
 
 __all__ = [
+    "extract_brick_imports",
+    "extract_brick_imports_with_namespaces",
     "extract_top_ns",
     "fetch_all_imports",
+    "fetch_api",
+    "fetch_brick_import_usages",
     "fetch_excluded_imports",
     "list_imports",
 ]

components/polylith/check/grouping.py → components/polylith/imports/grouping.py

@@ -34,3 +34,9 @@ def extract_brick_imports(all_imports: dict, top_ns) -> dict:
     with_only_brick_names = only_brick_names(with_only_bricks)
 
     return exclude_empty(with_only_brick_names)
+
+
+def extract_brick_imports_with_namespaces(all_imports: dict, top_ns) -> dict:
+    with_only_bricks = only_bricks(all_imports, top_ns)
+
+    return exclude_empty(with_only_bricks)

components/polylith/imports/parser.py

@@ -2,7 +2,7 @@
 from collections.abc import Iterable
 from functools import lru_cache
 from pathlib import Path
-from typing import List, Set, Union
+from typing import FrozenSet, List, Mapping, Optional, Set, Tuple, Union
 
 typing_ns = "typing"
 type_checking = "TYPE_CHECKING"
@@ -68,6 +68,56 @@ def parse_node(node: ast.AST) -> Union[dict, None]:
     return None
 
 
+def extract_api_part(path: str) -> str:
+    return path.rsplit(".", 1)[-1]
+
+
+def find_import_root_and_path(
+    expr: ast.expr, parts: Tuple[str, ...] = ()
+) -> Tuple[ast.expr, str]:
+    """Builds a namespace when the expression is an Attribute or Name, otherwise empty."""
+    if isinstance(expr, ast.Attribute):
+        return find_import_root_and_path(expr.value, (*parts, expr.attr))
+
+    namespace_parts = (*parts, expr.id) if isinstance(expr, ast.Name) else parts
+
+    namespace = str.join(".", reversed(namespace_parts))
+
+    return expr, namespace
+
+
+def find_matching_usage(expr: ast.expr, api_map: Mapping[str, str]) -> Optional[str]:
+    root, usage = find_import_root_and_path(expr)
+
+    if isinstance(root, ast.Name) and root.id in api_map:
+        return usage
+
+    return None
+
+
+def parse_import_usage(node: ast.AST, api_map: Mapping[str, str]) -> Union[str, None]:
+    usage = None
+    child = None
+
+    wrapper_nodes = (ast.Await, ast.Expr, ast.NamedExpr, ast.Starred, ast.Subscript)
+
+    if isinstance(node, ast.Attribute):
+        usage = find_matching_usage(node, api_map)
+        child = node.value
+    elif isinstance(node, wrapper_nodes):
+        child = node.value
+    elif isinstance(node, ast.Call):
+        usage = find_matching_usage(node.func, api_map)
+        child = node.func
+    elif isinstance(node, ast.UnaryOp):
+        child = node.operand
+
+    if usage:
+        return usage
+
+    return parse_import_usage(child, api_map) if child is not None else None
+
+
 def parse_module(path: Path) -> ast.AST:
     with open(path.as_posix(), "r", encoding="utf-8", errors="ignore") as f:
         tree = ast.parse(f.read(), path.name)
@@ -113,6 +163,36 @@ def fetch_all_imports(paths: Set[Path]) -> dict:
     return {k: v for row in rows for k, v in row.items()}
 
 
+def fetch_import_usages_in_module(path: Path, imported: FrozenSet[str]) -> Set[str]:
+    tree = parse_module(path)
+    api_map = {extract_api_part(p): p for p in imported}
+
+    nodes = (parse_import_usage(n, api_map) for n in ast.walk(tree))
+
+    return {n for n in nodes if n is not None}
+
+
+@lru_cache(maxsize=None)
+def fetch_brick_import_usages(path: Path, imported: FrozenSet[str]) -> Set[str]:
+    py_modules = find_files(path)
+
+    res = (fetch_import_usages_in_module(p, imported) for p in py_modules)
+
+    return {i for n in res if n for i in n}
+
+
+def extract_api(paths: Set[str]) -> Set[str]:
+    return {extract_api_part(p) for p in paths}
+
+
+def fetch_api(paths: Set[Path]) -> dict:
+    interfaces = [Path(p / "__init__.py") for p in paths]
+
+    rows = [{i.parent.name: extract_api(list_imports(i))} for i in interfaces]
+
+    return {k: v for row in rows for k, v in row.items()}
+
+
 def should_exclude(path: Path, excludes: Set[str]):
     return any(path.match(pattern) for pattern in excludes)
 

components/polylith/interface/__init__.py

@@ -1,3 +1,4 @@
+from polylith.interface import report
 from polylith.interface.interfaces import create_interface
 
-__all__ = ["create_interface"]
+__all__ = ["create_interface", "report"]

components/polylith/interface/report.py

@@ -0,0 +1,140 @@
+from pathlib import Path
+from typing import Dict, FrozenSet, Set, Tuple
+
+from polylith import imports
+from polylith.reporting import theme
+from polylith.workspace.paths import collect_bases_paths, collect_components_paths
+from rich.console import Console
+from rich.padding import Padding
+from rich.table import Table
+
+
+def get_brick_interface(root: Path, ns: str, brick: str, bricks: dict) -> set:
+    bases = bricks["bases"]
+    paths = {brick}
+
+    fn = collect_bases_paths if brick in bases else collect_components_paths
+
+    brick_paths = fn(root, ns, paths)
+    bricks_api = imports.fetch_api(brick_paths)
+    brick_api = bricks_api.get(brick) or set()
+
+    return {f"{ns}.{brick}.{endpoint}" for endpoint in brick_api}
+
+
+def get_brick_imports(root: Path, ns: str, bases: set, components: set) -> dict:
+    bases_paths = collect_bases_paths(root, ns, bases)
+    components_paths = collect_components_paths(root, ns, components)
+
+    in_bases = imports.fetch_all_imports(bases_paths)
+    in_comps = imports.fetch_all_imports(components_paths)
+
+    extracted_bases = imports.extract_brick_imports_with_namespaces(in_bases, ns)
+    extracted_components = imports.extract_brick_imports_with_namespaces(in_comps, ns)
+
+    return {**extracted_bases, **extracted_components}
+
+
+def to_imported_api(brick_imports: Set[str]) -> Set[str]:
+    return {imports.parser.extract_api_part(b) for b in brick_imports}
+
+
+def filter_by_brick(brick_imports: Set[str], brick: str, ns: str) -> Set[str]:
+    brick_with_ns = f"{ns}.{brick}"
+
+    return {b for b in brick_imports if str.startswith(b, brick_with_ns)}
+
+
+def is_matching_namespace(using: str, endpoint: str) -> bool:
+    return str.startswith(endpoint, using) or str.startswith(using, endpoint)
+
+
+def is_within_namespace(using: str, brick_interface: Set[str]) -> bool:
+    return any(is_matching_namespace(using, i) for i in brick_interface)
+
+
+def check_usage(usings: Set[str], brick_interface: Set[str]) -> dict:
+    return {u: is_within_namespace(u, brick_interface) for u in usings}
+
+
+def frozen(data: Dict[str, Set[str]], key: str) -> FrozenSet[str]:
+    return frozenset(data.get(key) or set())
+
+
+def check_brick_interface_usage(
+    root: Path, ns: str, brick: str, bricks: dict
+) -> Tuple[set, dict]:
+    brick_interface = get_brick_interface(root, ns, brick, bricks)
+
+    bases = bricks["bases"]
+    components = bricks["components"]
+
+    brick_imports = get_brick_imports(root, ns, bases, components)
+    by_brick = {k: filter_by_brick(v, brick, ns) for k, v in brick_imports.items()}
+
+    bases_paths = collect_bases_paths(root, ns, bases)
+    comp_paths = collect_components_paths(root, ns, components)
+    paths = bases_paths.union(comp_paths)
+
+    usage = {
+        p.name: imports.fetch_brick_import_usages(p, frozen(by_brick, p.name))
+        for p in paths
+    }
+
+    imports_and_usage = {k: {*v, *(by_brick.get(k) or set())} for k, v in usage.items()}
+    checked = {k: check_usage(v, brick_interface) for k, v in imports_and_usage.items()}
+
+    return brick_interface, checked
+
+
+def has_valid_usage(checked_usage: dict) -> bool:
+    return all(v for v in checked_usage.values())
+
+
+def print_brick_interface(brick: str, brick_interface: set, bricks: dict) -> None:
+    console = Console(theme=theme.poly_theme)
+
+    tag = "base" if brick in bricks["bases"] else "comp"
+
+    table = Table(box=None)
+
+    message = f"[{tag}]{brick}[/] exposes:"
+    table.add_column(Padding(message, (1, 0, 0, 0)))
+
+    for endpoint in sorted(brick_interface):
+        *_ns, exposes = str.split(endpoint, ".")
+        table.add_row(f"[data]{exposes}[/]")
+
+    console.print(table, overflow="ellipsis")
+
+
+def print_brick_interface_usage(root: Path, ns: str, brick: str, bricks: dict) -> None:
+    brick_interface, res = check_brick_interface_usage(root, ns, brick, bricks)
+
+    invalid_usage = {k: v for k, v in res.items() if not has_valid_usage(v)}
+
+    if not invalid_usage:
+        return
+
+    console = Console(theme=theme.poly_theme)
+
+    table = Table(box=None)
+    tag = "base" if brick in bricks["bases"] else "comp"
+
+    for using_brick, usages in invalid_usage.items():
+        using_tag = "base" if using_brick in bricks["bases"] else "comp"
+        usings = {k for k, v in usages.items() if v is False}
+
+        for using in usings:
+            _ns, _used_brick, *feature = str.split(using, ".")
+            used = str.join(".", feature)
+
+            prefix = f"In [{using_tag}]{using_brick}[/]"
+            middle = f"[{tag}]{brick}[/].[data]{used}[/] is not part of the public interface of [{tag}]{brick}[/]"
+            suffix = "as defined in [data]__init__.py[/]"
+
+            message = f":bulb: {prefix}, {middle} {suffix}."
+
+            table.add_row(f"{message}")
+
+    console.print(table, overflow="ellipsis")

components/polylith/test/core.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 from typing import List, Union
 
-from polylith import check, diff, imports
+from polylith import diff, imports
 
 
 def is_test(root: Path, ns: str, path: Path, theme: str) -> bool:
@@ -34,4 +34,4 @@ def get_brick_imports_in_tests(
 
     all_imports = {k: v for k, v in enumerate(listed_imports)}
 
-    return check.grouping.extract_brick_imports(all_imports, ns)
+    return imports.extract_brick_imports(all_imports, ns)