Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 17 additions & 11 deletions fastapi_startkit/src/fastapi_startkit/facades/Loader.pyi
Original file line number Diff line number Diff line change
@@ -1,26 +1,32 @@
from typing import Any
from typing import Any, Callable

class Loader:
def get_modules(files_or_directories: list, raise_exception: bool = False) -> dict:
@staticmethod
def get_modules(files_or_directories: str | list[str], raise_exception: bool = False) -> dict[str, Any]:
"""Get a list of Python modules found (recursively) in the given list of files or directories.
If raise_exception is enabled it will raise an exception in case of error during loading a module."""
@staticmethod
def find(
class_instance: Any,
paths: list,
class_instance: type,
paths: str | list[str],
class_name: str,
raise_exception: bool = False,
) -> "None|Any": ...
def find_all(class_instance: Any, paths: list, raise_exception: bool = False) -> dict: ...
def get_object(path_or_module: "str|Any", object_name: str, raise_exception: bool = False) -> Any:
) -> type | None: ...
@staticmethod
def find_all(class_instance: type, paths: str | list[str], raise_exception: bool = False) -> dict[str, type]: ...
@staticmethod
def get_object(path_or_module: str | Any, object_name: str, raise_exception: bool = False) -> Any:
"""Load the given object from a Python module located at path and returns a default value if
not found. If no object name is provided, returns the loaded module."""
...
@staticmethod
def get_objects(
path_or_module: "str|Any",
filter_method: callable = None,
path_or_module: str | Any,
filter_method: Callable[[Any], bool] | None = None,
raise_exception: bool = False,
) -> dict:
) -> dict[str, Any] | None:
"""Returns a dictionary of objects from the given path (file or dotted). The dictionary can
be filtered if a given callable is given."""
...
def get_parameters(module_or_path: "str|Any") -> dict: ...
@staticmethod
def get_parameters(module_or_path: str | Any) -> dict[str, Any]: ...
36 changes: 23 additions & 13 deletions fastapi_startkit/src/fastapi_startkit/facades/RateLimiter.pyi
Original file line number Diff line number Diff line change
@@ -1,41 +1,51 @@
from typing import Any, Callable, TYPE_CHECKING
from typing import Any, Callable

if TYPE_CHECKING:
from ..rates.limiters import Limiter
Limiter = Any

class RateLimiter:
"""Rate Limiter facades to add rate limiting to your functions."""

def register(self, name, callback: "Limiter") -> "RateLimiter":
@staticmethod
def register(name: str, callback: "Limiter") -> "RateLimiter":
"""Register a new rate limiter with the given name"""
...
def attempts(self, key: str) -> int:
@staticmethod
def attempts(key: str) -> int:
"""Get number of attempts left for a given rate limiter key."""
...
def get_limiter(self, name: str) -> "Limiter":
@staticmethod
def get_limiter(name: str) -> "Limiter":
"""Get rate limiter registered with the given name."""
...
@staticmethod
def attempt(key: str, callback: Callable, max_attempts: int, delay: int = 60) -> Any:
"""Try to execute the given callback if not limited by the 'key' rate limiter."""
...
def too_many_attempts(self, key: str, max_attempts: int) -> bool:
@staticmethod
def too_many_attempts(key: str, max_attempts: int) -> bool:
"""Check if given rate limiter key got more (or equal) attempts than max_attempts."""
...
def hit(self, key: str, delay: int) -> int:
@staticmethod
def hit(key: str, delay: int) -> int:
"""Add one attempt for the given key."""
...
def reset_attempts(self, key: str) -> bool:
@staticmethod
def reset_attempts(key: str) -> bool:
"""Reset attempts count to 0 for the given key."""
...
def clear(self, key: str):
@staticmethod
def clear(key: str):
"""Clear all data of the given rate limiter key."""
...
def available_at(self, key: str) -> int:
@staticmethod
def available_at(key: str) -> int:
"""Get UNIX integer timestamp at which rate limiter key will be available again."""
...
def available_in(self, key: str) -> int:
@staticmethod
def available_in(key: str) -> int:
"""Get seconds in which rate limiter key will be available again."""
...
def remaining(self, key: str, max_attempts: int) -> int:
@staticmethod
def remaining(key: str, max_attempts: int) -> int:
"""Get remaining attempts before given rate limiter key is limited regarding max_attempts limit."""
...
7 changes: 6 additions & 1 deletion fastapi_startkit/src/fastapi_startkit/facades/Url.pyi
Original file line number Diff line number Diff line change
@@ -1,18 +1,23 @@
from typing import Any

class Url:
"""URL helper facade."""

@staticmethod
def url(path: str = "") -> str:
"""Generates a fully qualified url to the given path. If no path is given this will return
the base url domain."""
...
@staticmethod
def asset(alias: str, filename: str) -> str:
"""Generates a fully qualified URL for the given asset using the given disk
Example:
asset("local", "avatar.jpg") (take first pat)
asset("s3.private", "doc.pdf") (when multiple paths are specified for the disk)
"""
...
def route(name: str, params: dict = {}, absolute: bool = True) -> str:
@staticmethod
def route(name: str, params: dict[str, Any] | None = None, absolute: bool = True) -> str:
"""Generates a fully qualified URL to the given route name.
Example:
route("users.home") : http://masonite.app/dashboard/
Expand Down
10 changes: 7 additions & 3 deletions fastapi_startkit/src/fastapi_startkit/loader/Loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import inspect
import pkgutil
from importlib.machinery import FileFinder

from ..exceptions import LoaderNotFound
from ..support.structures import load
Expand All @@ -18,9 +19,12 @@ def get_modules(self, files_or_directories, raise_exception=False):

_modules = {}
module_paths = list(map(lambda p: p.replace(".", "/"), files_or_directories))
for module_loader, name, _ in pkgutil.iter_modules(module_paths):
for module_finder, name, _ in pkgutil.iter_modules(module_paths):
# load() imports from a file path, so only filesystem finders (not e.g. zip archives) apply.
if not isinstance(module_finder, FileFinder):
continue
module = load(
f"{module_loader.path}/{name}.py",
f"{module_finder.path}/{name}.py",
raise_exception=raise_exception,
)
_modules.update({name: module})
Expand Down Expand Up @@ -64,7 +68,7 @@ def get_objects(self, path_or_module, filter_method=None, raise_exception=False)

def get_parameters(self, module_or_path):
_parameters = {}
for name, obj in self.get_objects(module_or_path).items():
for name, obj in (self.get_objects(module_or_path) or {}).items():
if parameters_filter(name, obj):
_parameters.update({name: obj})

Expand Down
12 changes: 12 additions & 0 deletions fastapi_startkit/tests/utils/test_loader.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import zipfile

import pytest

from fastapi_startkit.exceptions.exceptions import LoaderNotFound
Expand Down Expand Up @@ -48,6 +50,13 @@ def test_accepts_single_path_or_list(self, module_dir):
loader = Loader()
assert loader.get_modules([module_dir]).keys() == loader.get_modules(module_dir).keys()

def test_skips_modules_not_found_on_the_filesystem(self, tmp_path):
archive = tmp_path / "archive"
with zipfile.ZipFile(archive, "w") as zf:
zf.writestr("zipped.py", MODULE_SOURCE)

assert Loader().get_modules(str(archive)) == {}


class TestFind:
def test_find_returns_matching_class(self, module_dir):
Expand Down Expand Up @@ -121,6 +130,9 @@ def test_returns_non_dunder_members(self, module_file):
assert "Dog" in params
assert not any(name.startswith("__") for name in params)

def test_missing_module_returns_empty(self, tmp_path):
assert Loader().get_parameters(f"{tmp_path}/does_not_exist.py") == {}


class TestParametersFilter:
def test_rejects_dunder(self):
Expand Down
Loading