Skip to content

Commit

Permalink
Typing fixes for combine/layer
Browse files Browse the repository at this point in the history
  • Loading branch information
lowell80 committed Oct 30, 2023
1 parent a2236fe commit c5f5125
Show file tree
Hide file tree
Showing 2 changed files with 15 additions and 16 deletions.
7 changes: 3 additions & 4 deletions ksconf/combine.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
import sys
from os import fspath
from pathlib import Path
from typing import Callable, Optional, Sequence
from typing import Callable, Optional, Sequence, Union

from ksconf.command import ConfFileProxy
from ksconf.compat import List, Tuple
Expand Down Expand Up @@ -123,7 +123,7 @@ def set_layer_root(self, root: LayerRootBase.Layer):
def add_layer_filter(self, action, pattern):
self.layer_filter.add_rule(action, pattern)

def combine(self, target: Path, *, hook_label=""):
def combine(self, target: Union[Path, str], *, hook_label=""):
"""
Combine layers into ``target`` directory.
Any ``hook_label`` given will be passed to the plugin system via the
Expand All @@ -132,6 +132,7 @@ def combine(self, target: Path, *, hook_label=""):
layer_root = self.layer_root
if layer_root is None:
raise TypeError("Call either set_source_dirs() or set_layer_root() before calling combine()")
target = Path(target)
self.prepare(target)
# Build a common tree of all src files.
src_file_listing = layer_root.list_files()
Expand All @@ -145,8 +146,6 @@ def prepare(self, target: Path):
""" Start the combine process. This includes directory checking,
applying layer filtering, and marker file handling. """
layer_root, layer_filter = self.layer_root, self.layer_filter
target = Path(target)

self.prepare_target_dir(target)

self.layer_names_all.update(layer_root.list_layer_names())
Expand Down
24 changes: 12 additions & 12 deletions ksconf/layer.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
from os import PathLike, stat_result
from pathlib import Path, PurePath
from tempfile import NamedTemporaryFile
from typing import Callable, Iterator, Match, Optional
from typing import Callable, Iterator, Optional, Pattern, Type, Union

from ksconf.compat import Dict, List, Set, Tuple
from ksconf.hook import plugin_manager
Expand Down Expand Up @@ -48,7 +48,7 @@
"""


def _path_join(*parts):
def _path_join(*parts) -> Path:
""" A slightly smarter / more flexible path appender.
Drop any None
"""
Expand All @@ -68,15 +68,15 @@ class LayerUsageException(LayerException):
@dataclass
class LayerContext:
follow_symlink: bool = False
block_files: Match = re.compile(r"\.(bak|swp)$")
block_files: Pattern = re.compile(r"\.(bak|swp)$")
block_dirs: set = field(default_factory=lambda: {".git"})
template_variables: dict = field(default_factory=dict)


@dataclass
class _FileFactoryHandler:
name: str
handler: LayerFile
handler: Type[LayerFile]
priority: int = 0
enabled: bool = False

Expand All @@ -87,7 +87,7 @@ class FileFactory:

def __init__(self):
# Make this class a singleton?
self._context_state = None
self._context_state: tuple = ()

def _recalculate(self):
self._enabled_handlers = [
Expand All @@ -105,7 +105,7 @@ def disable(self, name):
def list_available_handlers(self) -> List[str]:
return [h.name for h in self._registered_handlers.values() if not h.enabled]

def __call__(self, layer, path: PurePath, *args, **kwargs) -> LayerFile:
def __call__(self, layer, path: PurePath, *args, **kwargs) -> Union[LayerFile, None]:
"""
Factory thats finds the appropriate LayerFile class and returns a new instance.
"""
Expand All @@ -114,7 +114,7 @@ def __call__(self, layer, path: PurePath, *args, **kwargs) -> LayerFile:
return cls(layer, path, *args, **kwargs)

def register_handler(self, name: str, **kwargs):
def wrapper(handler_class: LayerFile):
def wrapper(handler_class: Type[LayerFile]) -> Type[LayerFile]:
handler = _FileFactoryHandler(name, handler_class, **kwargs)
self._registered_handlers[handler.name] = handler
self._recalculate()
Expand All @@ -131,7 +131,7 @@ def __enter__(self) -> FileFactory:

def __exit__(self, exc_type, exc_val, exc_tb):
self._registered_handlers, self._enabled_handlers = self._context_state
self._context_state = None
self._context_state = ()


# Single shared instance
Expand Down Expand Up @@ -173,7 +173,7 @@ def __init__(self,
self._stat = stat

def __fspath__(self) -> str:
return self.resource_path
return str(self.resource_path)

@staticmethod
def match(path: PurePath):
Expand Down Expand Up @@ -216,7 +216,7 @@ class LayerRenderedFile(LayerFile):

def __init__(self, *args, **kwargs):
super(LayerRenderedFile, self).__init__(*args, **kwargs)
self._rendered_resource = None
self._rendered_resource: Path = None

def __del__(self):
if getattr(self, "_rendered_resource", None) and self._rendered_resource.is_file():
Expand Down Expand Up @@ -307,7 +307,7 @@ def add_rule(self, action, pattern):
if not self._rules:
if action == "include":
first_filter = ("exclude", "*")
elif "exclude":
else:
first_filter = ("include", "*")
self._rules.append(first_filter)
self._rules.append((action, pattern))
Expand Down Expand Up @@ -369,7 +369,7 @@ def list_files(self) -> List[LayerFile]:
self._cache_files = list(self.iter_files())
return self._cache_files

def get_file(self, path: Path) -> LayerFile:
def get_file(self, path: Path) -> Union[LayerFile, None]:
""" Return file object (by logical path), if it exists in this layer. """
# TODO: Optimize by caching. Use a dict with a logical_path as the key
for file in self.list_files():
Expand Down

0 comments on commit c5f5125

Please sign in to comment.