feat: [CG-10650] codebase.codeowners interface

ruff.toml

@@ -57,6 +57,7 @@ extend-generics = [
   "codegen.sdk.core.assignment.Assignment",
   "codegen.sdk.core.class_definition.Class",
   "codegen.sdk.core.codebase.Codebase",
+  "codegen.sdk.core.codeowner.CodeOwner",
   "codegen.sdk.core.dataclasses.usage.Usage",
   "codegen.sdk.core.dataclasses.usage.UsageType",
   "codegen.sdk.core.dataclasses.usage.UsageKind",

src/codegen/sdk/core/codebase.py

@@ -7,6 +7,7 @@
 import re
 from collections.abc import Generator
 from contextlib import contextmanager
+from functools import cached_property
 from pathlib import Path
 from typing import TYPE_CHECKING, Generic, Literal, TypeVar, Unpack, overload
 
@@ -36,6 +37,7 @@
 from codegen.sdk.codebase.span import Span
 from codegen.sdk.core.assignment import Assignment
 from codegen.sdk.core.class_definition import Class
+from codegen.sdk.core.codeowner import CodeOwner
 from codegen.sdk.core.detached_symbols.code_block import CodeBlock
 from codegen.sdk.core.detached_symbols.parameter import Parameter
 from codegen.sdk.core.directory import Directory
@@ -257,6 +259,17 @@ def files(self, *, extensions: list[str] | Literal["*"] | None = None) -> list[T
         # Sort files alphabetically
         return sort_editables(files, alphabetical=True, dedupe=False)
 
+    @cached_property
+    def codeowners(self) -> list["CodeOwner[TSourceFile]"]:
+        """List all CodeOnwers in the codebase.
+
+        Returns:
+            list[CodeOwners]: A list of CodeOwners objects in the codebase.
+        """
+        if self.G.codeowners_parser is None:
+            return []
+        return CodeOwner.from_parser(self.G.codeowners_parser, lambda *args, **kwargs: self.files(*args, **kwargs))
+
     @property
     def directories(self) -> list[TDirectory]:
         """List all directories in the codebase.

src/codegen/sdk/core/codeowner.py

@@ -0,0 +1,97 @@
+import logging
+from collections.abc import Iterable, Iterator
+from typing import Callable, Generic, Literal
+
+from codeowners import CodeOwners as CodeOwnersParser
+
+from codegen.sdk._proxy import proxy_property
+from codegen.sdk.core.interfaces.has_symbols import (
+    FilesParam,
+    HasSymbols,
+    TClass,
+    TFile,
+    TFunction,
+    TGlobalVar,
+    TImport,
+    TImportStatement,
+    TSymbol,
+)
+from codegen.sdk.core.utils.cache_utils import cached_generator
+from codegen.shared.decorators.docs import apidoc, noapidoc
+
+logger = logging.getLogger(__name__)
+
+
+@apidoc
+class CodeOwner(
+    HasSymbols[TFile, TSymbol, TImportStatement, TGlobalVar, TClass, TFunction, TImport],
+    Generic[TFile, TSymbol, TImportStatement, TGlobalVar, TClass, TFunction, TImport],
+):
+    """CodeOwner is a class that represents a code owner in a codebase.
+
+    It is used to iterate over all files that are owned by a specific owner.
+
+    Attributes:
+        owner_type: The type of the owner (USERNAME, TEAM, EMAIL).
+        owner_value: The value of the owner.
+        files_source: A callable that returns an iterable of all files in the codebase.
+    """
+
+    owner_type: Literal["USERNAME", "TEAM", "EMAIL"]
+    owner_value: str
+    files_source: Callable[FilesParam, Iterable[TFile]]
+
+    def __init__(
+        self,
+        files_source: Callable[FilesParam, Iterable[TFile]],
+        owner_type: Literal["USERNAME", "TEAM", "EMAIL"],
+        owner_value: str,
+    ):
+        self.owner_type = owner_type
+        self.owner_value = owner_value
+        self.files_source = files_source
+
+    @classmethod
+    def from_parser(
+        cls,
+        parser: CodeOwnersParser,
+        file_source: Callable[FilesParam, Iterable[TFile]],
+    ) -> list["CodeOwner"]:
+        """Create a list of CodeOwner objects from a CodeOwnersParser.
+
+        Args:
+            parser (CodeOwnersParser): The CodeOwnersParser to use.
+            file_source (Callable[FilesParam, Iterable[TFile]]): A callable that returns an iterable of all files in the codebase.
+
+        Returns:
+            list[CodeOwner]: A list of CodeOwner objects.
+        """
+        codeowners = []
+        for _, _, owners, _, _ in parser.paths:
+            for owner_label, owner_value in owners:
+                codeowners.append(CodeOwner(file_source, owner_label, owner_value))
+        return codeowners
+
+    @cached_generator(maxsize=16)
+    @noapidoc
+    def files_generator(self, *args: FilesParam.args, **kwargs: FilesParam.kwargs) -> Iterable[TFile]:
+        for source_file in self.files_source(*args, **kwargs):
+            # Filter files by owner value
+            if self.owner_value in source_file.owners:
+                yield source_file
+
+    @proxy_property
+    def files(self, *args: FilesParam.args, **kwargs: FilesParam.kwargs) -> Iterable[TFile]:
+        """Recursively iterate over all files in the codebase that are owned by the current code owner."""
+        return self.files_generator(*args, **kwargs)
+
+    @property
+    def name(self) -> str:
+        """The name of the code owner."""
+        return self.owner_value
+
+    def __iter__(self) -> Iterator[TFile]:
+        return iter(self.files_generator())
+
+    def __repr__(self) -> str:
+        return f"CodeOwner(owner_type={self.owner_type}, owner_value={self.owner_value})"

src/codegen/sdk/core/directory.py

@@ -1,43 +1,30 @@
+import logging
 import os
-from itertools import chain
+from collections.abc import Iterator
 from pathlib import Path
-from typing import TYPE_CHECKING, Generic, Self, TypeVar
-
-from codegen.shared.decorators.docs import apidoc, py_noapidoc
-
-if TYPE_CHECKING:
-    from codegen.sdk.core.assignment import Assignment
-    from codegen.sdk.core.class_definition import Class
-    from codegen.sdk.core.file import File
-    from codegen.sdk.core.function import Function
-    from codegen.sdk.core.import_resolution import Import, ImportStatement
-    from codegen.sdk.core.symbol import Symbol
-    from codegen.sdk.typescript.class_definition import TSClass
-    from codegen.sdk.typescript.export import TSExport
-    from codegen.sdk.typescript.file import TSFile
-    from codegen.sdk.typescript.function import TSFunction
-    from codegen.sdk.typescript.import_resolution import TSImport
-    from codegen.sdk.typescript.statements.import_statement import TSImportStatement
-    from codegen.sdk.typescript.symbol import TSSymbol
-
-import logging
+from typing import Generic, Self
+
+from codegen.sdk.core.interfaces.has_symbols import (
+    HasSymbols,
+    TClass,
+    TFile,
+    TFunction,
+    TGlobalVar,
+    TImport,
+    TImportStatement,
+    TSymbol,
+)
+from codegen.sdk.core.utils.cache_utils import cached_generator
+from codegen.shared.decorators.docs import apidoc, noapidoc
 
 logger = logging.getLogger(__name__)
 
 
-TFile = TypeVar("TFile", bound="File")
-TSymbol = TypeVar("TSymbol", bound="Symbol")
-TImportStatement = TypeVar("TImportStatement", bound="ImportStatement")
-TGlobalVar = TypeVar("TGlobalVar", bound="Assignment")
-TClass = TypeVar("TClass", bound="Class")
-TFunction = TypeVar("TFunction", bound="Function")
-TImport = TypeVar("TImport", bound="Import")
-
-TSGlobalVar = TypeVar("TSGlobalVar", bound="Assignment")
-
-
 @apidoc
-class Directory(Generic[TFile, TSymbol, TImportStatement, TGlobalVar, TClass, TFunction, TImport]):
+class Directory(
+    HasSymbols[TFile, TSymbol, TImportStatement, TGlobalVar, TClass, TFunction, TImport],
+    Generic[TFile, TSymbol, TImportStatement, TGlobalVar, TClass, TFunction, TImport],
+):
     """Directory representation for codebase.
 
     GraphSitter abstraction of a file directory that can be used to look for files and symbols within a specific directory.
@@ -58,7 +45,7 @@ def __init__(self, path: Path, dirpath: str, parent: Self | None):
         self.path = path
         self.dirpath = dirpath
         self.parent = parent
-        self.items = dict()
+        self.items = {}
 
     def __iter__(self):
         return iter(self.items.values())
@@ -126,62 +113,13 @@ def _get_subdirectories(directory: Directory):
         _get_subdirectories(self)
         return subdirectories
 
-    @property
-    def symbols(self) -> list[TSymbol]:
-        """Get a recursive list of all symbols in the directory and its subdirectories."""
-        return list(chain.from_iterable(f.symbols for f in self.files))
-
-    @property
-    def import_statements(self) -> list[TImportStatement]:
-        """Get a recursive list of all import statements in the directory and its subdirectories."""
-        return list(chain.from_iterable(f.import_statements for f in self.files))
-
-    @property
-    def global_vars(self) -> list[TGlobalVar]:
-        """Get a recursive list of all global variables in the directory and its subdirectories."""
-        return list(chain.from_iterable(f.global_vars for f in self.files))
-
-    @property
-    def classes(self) -> list[TClass]:
-        """Get a recursive list of all classes in the directory and its subdirectories."""
-        return list(chain.from_iterable(f.classes for f in self.files))
-
-    @property
-    def functions(self) -> list[TFunction]:
-        """Get a recursive list of all functions in the directory and its subdirectories."""
-        return list(chain.from_iterable(f.functions for f in self.files))
-
-    @property
-    @py_noapidoc
-    def exports(self: "Directory[TSFile, TSSymbol, TSImportStatement, TSGlobalVar, TSClass, TSFunction, TSImport]") -> "list[TSExport]":
-        """Get a recursive list of all exports in the directory and its subdirectories."""
-        return list(chain.from_iterable(f.exports for f in self.files))
-
-    @property
-    def imports(self) -> list[TImport]:
-        """Get a recursive list of all imports in the directory and its subdirectories."""
-        return list(chain.from_iterable(f.imports for f in self.files))
-
-    def get_symbol(self, name: str) -> TSymbol | None:
-        """Get a symbol by name in the directory and its subdirectories."""
-        return next((s for s in self.symbols if s.name == name), None)
-
-    def get_import_statement(self, name: str) -> TImportStatement | None:
-        """Get an import statement by name in the directory and its subdirectories."""
-        return next((s for s in self.import_statements if s.name == name), None)
-
-    def get_global_var(self, name: str) -> TGlobalVar | None:
-        """Get a global variable by name in the directory and its subdirectories."""
-        return next((s for s in self.global_vars if s.name == name), None)
-
-    def get_class(self, name: str) -> TClass | None:
-        """Get a class by name in the directory and its subdirectories."""
-        return next((s for s in self.classes if s.name == name), None)
-
-    def get_function(self, name: str) -> TFunction | None:
-        """Get a function by name in the directory and its subdirectories."""
-        return next((s for s in self.functions if s.name == name), None)
+    @noapidoc
+    @cached_generator()
+    def files_generator(self) -> Iterator[TFile]:
+        """Yield files recursively from the directory."""
+        yield from self.files
 
+    # Directory-specific methods
     def add_file(self, file: TFile) -> None:
         """Add a file to the directory."""
         rel_path = os.path.relpath(file.file_path, self.dirpath)
@@ -202,18 +140,12 @@ def get_file(self, filename: str, ignore_case: bool = False) -> TFile | None:
         from codegen.sdk.core.file import File
 
         if ignore_case:
-            return next((f for name, f in self.items.items() if name.lower() == filename.lower() and isinstance(f, File)), None)
+            return next(
+                (f for name, f in self.items.items() if name.lower() == filename.lower() and isinstance(f, File)),
+                None,
+            )
         return self.items.get(filename, None)
 
-    @py_noapidoc
-    def get_export(self: "Directory[TSFile, TSSymbol, TSImportStatement, TSGlobalVar, TSClass, TSFunction, TSImport]", name: str) -> "TSExport | None":
-        """Get an export by name in the directory and its subdirectories (supports only typescript)."""
-        return next((s for s in self.exports if s.name == name), None)
-
-    def get_import(self, name: str) -> TImport | None:
-        """Get an import by name in the directory and its subdirectories."""
-        return next((s for s in self.imports if s.name == name), None)
-
     def add_subdirectory(self, subdirectory: Self) -> None:
         """Add a subdirectory to the directory."""
         rel_path = os.path.relpath(subdirectory.dirpath, self.dirpath)
@@ -230,23 +162,22 @@ def remove_subdirectory_by_path(self, subdirectory_path: str) -> None:
         del self.items[rel_path]
 
     def get_subdirectory(self, subdirectory_name: str) -> Self | None:
-        """Get a subdirectory by its path relative to the directory."""
+        """Get a subdirectory by its name (relative to the directory)."""
         return self.items.get(subdirectory_name, None)
 
-    def remove(self) -> None:
-        """Remove the directory and all its files and subdirectories."""
-        for f in self.files:
-            f.remove()
-
     def update_filepath(self, new_filepath: str) -> None:
-        """Update the filepath of the directory."""
+        """Update the filepath of the directory and its contained files."""
         old_path = self.dirpath
         new_path = new_filepath
-
         for file in self.files:
             new_file_path = os.path.join(new_path, os.path.relpath(file.file_path, old_path))
             file.update_filepath(new_file_path)
 
+    def remove(self) -> None:
+        """Remove all the files in the files container."""
+        for f in self.files:
+            f.remove()
+
     def rename(self, new_name: str) -> None:
         """Rename the directory."""
         parent_dir, _ = os.path.split(self.dirpath)