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
 
@@ -35,6 +36,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
@@ -256,6 +258,17 @@
         # 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,68 @@
+import logging
+from collections.abc import Iterable, Iterator
+from typing import Callable, Generic, Literal
+
+from codeowners import CodeOwners as CodeOwnersParser
+
+from codegen.sdk.core.interfaces.files_interface import FilesInterface, FilesParam, TClass, TFile, TFunction, TGlobalVar, TImport, TImportStatement, TSymbol
+from codegen.sdk.core.utils.cache_utils import cached_generator
+from codegen.shared.decorators.docs import apidoc
+
+logger = logging.getLogger(__name__)
+
+
+@apidoc
+class CodeOwner(FilesInterface[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
+        self.files = self.files_generator
+
+    @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)
+    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
+
+    @property
+    def name(self) -> str:
+        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,18 @@
+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
+from typing import Generic, Self
 
-import logging
+from codegen.sdk.core.interfaces.files_interface import FilesInterface, TClass, TFile, TFunction, TGlobalVar, TImport, TImportStatement, TSymbol
+from codegen.sdk.core.utils.cache_utils import cached_generator
+from codegen.shared.decorators.docs import apidoc
 
 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(FilesInterface[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 +33,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 +101,12 @@ 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)
+    @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)
@@ -205,15 +130,6 @@ def get_file(self, filename: str, ignore_case: bool = False) -> TFile | 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 +146,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)

src/codegen/sdk/core/interfaces/files_interface.py

@@ -0,0 +1,113 @@
+import logging
+from collections.abc import Iterator
+from itertools import chain
+from typing import TYPE_CHECKING, Generic, ParamSpec, TypeVar
+
+from codegen.sdk.core.utils.cache_utils import cached_generator
+from codegen.shared.decorators.docs import 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 SourceFile
+    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
+
+logger = logging.getLogger(__name__)
+
+
+TFile = TypeVar("TFile", bound="SourceFile")
+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")
+FilesParam = ParamSpec("FilesParam")
+
+TSGlobalVar = TypeVar("TSGlobalVar", bound="Assignment")
+
+
+class FilesInterface(Generic[TFile, TSymbol, TImportStatement, TGlobalVar, TClass, TFunction, TImport]):
+    """Abstract interface for files in a codebase.
+
+    Abstract interface for files in a codebase.
+    """
+
+    @cached_generator()
+    def files_generator(self, *args: FilesParam.args, **kwargs: FilesParam.kwargs) -> Iterator[TFile]:
+        msg = "This method should be implemented by the subclass"
+        raise NotImplementedError(msg)
+
+    @property
+    def symbols(self) -> list[TSymbol]:
+        """Get a recursive list of all symbols in files container."""
+        return list(chain.from_iterable(f.symbols for f in self.files_generator()))
+
+    @property
+    def import_statements(self) -> list[TImportStatement]:
+        """Get a recursive list of all import statements in files container."""
+        return list(chain.from_iterable(f.import_statements for f in self.files_generator()))
+
+    @property
+    def global_vars(self) -> list[TGlobalVar]:
+        """Get a recursive list of all global variables in files container."""
+        return list(chain.from_iterable(f.global_vars for f in self.files_generator()))
+
+    @property
+    def classes(self) -> list[TClass]:
+        """Get a recursive list of all classes in files container."""
+        return list(chain.from_iterable(f.classes for f in self.files_generator()))
+
+    @property
+    def functions(self) -> list[TFunction]:
+        """Get a recursive list of all functions in files container."""
+        return list(chain.from_iterable(f.functions for f in self.files_generator()))
+
+    @property
+    @py_noapidoc
+    def exports(self) -> "list[TSExport]":
+        """Get a recursive list of all exports in files container."""
+        return list(chain.from_iterable(f.exports for f in self.files_generator()))
+
+    @property
+    def imports(self) -> list[TImport]:
+        """Get a recursive list of all imports in files container."""
+        return list(chain.from_iterable(f.imports for f in self.files_generator()))
+
+    def get_symbol(self, name: str) -> TSymbol | None:
+        """Get a symbol by name in files container."""
+        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 files container."""
+        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 files container."""
+        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 files container."""
+        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 files container."""
+        return next((s for s in self.functions if s.name == name), None)
+
+    @py_noapidoc
+    def get_export(self: "FilesInterface[TSFile, TSSymbol, TSImportStatement, TSGlobalVar, TSClass, TSFunction, TSImport]", name: str) -> "TSExport | None":
+        """Get an export by name in files container (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 files container."""
+        return next((s for s in self.imports if s.name == name), None)