Skip to content

FastAuth utils

Password helper

fastauth.utils.password

PasswordHelperProtocol

Bases: Protocol

verify_and_update 
verify_and_update(plain_password, hashed_password)
Source code in fastauth/utils/password.py 
10
11
12
def verify_and_update(
    self, plain_password: str, hashed_password: str
) -> tuple[bool, str | None]: ...  # pragma: no cover
hash 
hash(password)
Source code in fastauth/utils/password.py 
14
def hash(self, password: str) -> str: ...  # pragma: no cover
generate 
generate()
Source code in fastauth/utils/password.py 
16
def generate(self) -> str: ...  # pragma: no cover

PasswordHelper 

PasswordHelper(password_hash=None)

Bases: PasswordHelperProtocol

Source code in fastauth/utils/password.py 
20
21
22
23
24
25
26
27
28
29
def __init__(self, password_hash: PasswordHash | None = None) -> None:
    if password_hash is None:
        self.password_hash = PasswordHash(
            (
                Argon2Hasher(),
                BcryptHasher(),
            )
        )
    else:
        self.password_hash = password_hash  # pragma: no cover
password_hash instance-attribute 
password_hash = PasswordHash((Argon2Hasher(), BcryptHasher()))
verify_and_update 
verify_and_update(plain_password, hashed_password)
Source code in fastauth/utils/password.py 
31
32
33
34
def verify_and_update(
    self, plain_password: str, hashed_password: str
) -> tuple[bool, str | None]:
    return self.password_hash.verify_and_update(plain_password, hashed_password)
hash 
hash(password)
Source code in fastauth/utils/password.py 
36
37
def hash(self, password: str) -> str:
    return self.password_hash.hash(password)
generate 
generate()
Source code in fastauth/utils/password.py 
39
40
def generate(self) -> str:
    return secrets.token_urlsafe()

JWT Decoder/Encoder

fastauth.utils.jwt_helper

TokenHelperProtocol

Bases: Protocol

decode_token 
decode_token(token, *args, **kwargs)
Source code in fastauth/utils/jwt_helper.py 
10
11
def decode_token(self, token: str, *args, **kwargs) -> dict[str, Any]:
    raise NotImplementedError
encode_token 
encode_token(payload, token_type, *args, **kwargs)
Source code in fastauth/utils/jwt_helper.py 
13
14
15
16
def encode_token(
    self, payload: dict[str, Any], token_type: TokenType, *args, **kwargs
) -> str:
    raise NotImplementedError

JWTHelper 

JWTHelper(secretkey, algorithm)
Source code in fastauth/utils/jwt_helper.py 
20
21
22
def __init__(self, secretkey: str, algorithm: str):
    self._secretkey = secretkey
    self._algorithm = algorithm
decode_token 
decode_token(token, audience=None, **kwargs)
Source code in fastauth/utils/jwt_helper.py 
24
25
26
27
28
29
30
31
32
33
def decode_token(
    self, token: str, audience: StringOrSequence | None = None, **kwargs
):
    return jwt.decode(
        token,
        key=self._secretkey,
        algorithms=[self._algorithm],
        audience=audience,
        **kwargs,
    )
encode_token 
encode_token(payload, token_type, max_age=None, audience=None, headers=None, **kwargs)
Source code in fastauth/utils/jwt_helper.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def encode_token(
    self,
    payload: dict[str, Any],
    token_type: TokenType,
    max_age: int | None = None,
    audience: StringOrSequence | None = None,
    headers: dict[str, Any] | None = None,
    **kwargs,
):
    payload["type"] = payload.get("type", token_type)
    payload["aud"] = payload.get("aud", audience)
    payload["iat"] = payload.get("iat", datetime.now(timezone.utc))
    payload["exp"] = payload.get(
        "exp", payload.get("iat") + timedelta(seconds=max_age)
    )
    return jwt.encode(
        payload,
        key=self._secretkey,
        algorithm=self._algorithm,
        headers=headers,
        **kwargs,
    )

Resolver for FastAPI DI system

fastauth.utils.injector

logger module-attribute 

logger = getLogger(__name__)

T module-attribute 

T = TypeVar('T')

DependencyError

Bases: Exception

Exception raised for errors during dependency injection.

injectable 

injectable(func=None, *, use_cache=True)

A decorator to enable FastAPI-style dependency injection for any function (sync or async).

This allows dependencies defined with FastAPI's Depends mechanism to be automatically resolved and injected into CLI tools or other components, not just web endpoints.

PARAMETER DESCRIPTION
func

The function to be wrapped, enabling dependency injection.

TYPE: Callable[..., T] | Callable[..., Coroutine[Any, Any, T]] | None DEFAULT: None

use_cache

Whether to use the dependency cache for the arguments a.k.a sub-dependencies.

TYPE: bool DEFAULT: True

RAISES DESCRIPTION
ValueError

If the dependant.call is not a callable function.
DependencyError

If an error occurs during dependency resolution.
Source code in fastauth/utils/injector.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def injectable(
    func: Callable[..., T] | Callable[..., Coroutine[Any, Any, T]] | None = None,
    *,
    use_cache: bool = True,
) -> Callable[..., T] | Callable[..., Coroutine[Any, Any, T]]:
    """A decorator to enable FastAPI-style dependency injection for any function (sync or async).

    This allows dependencies defined with FastAPI's Depends mechanism to be automatically
    resolved and injected into CLI tools or other components, not just web endpoints.

    :param func: The function to be wrapped, enabling dependency injection.
    :param use_cache: Whether to use the dependency cache for the arguments a.k.a sub-dependencies.
    :raise ValueError: If the dependant.call is not a callable function.
    :raise DependencyError: If an error occurs during dependency resolution.
    :returns The wrapped function with dependencies injected.
    """

    def _impl(
        func: Callable[..., T] | Callable[..., Coroutine[Any, Any, T]],
    ) -> Callable[..., T] | Callable[..., Coroutine[Any, Any, T]]:
        is_async = inspect.iscoroutinefunction(func)
        dependency_cache = _SOLVED_DEPENDENCIES if use_cache else None

        async def resolve_dependencies(
            dependant: Dependant,
        ) -> tuple[dict[str, Any], list[Any] | None]:
            fake_request = Request({"type": "http", "headers": [], "query_string": ""})
            async with AsyncExitStack() as stack:
                solved_result = await solve_dependencies(
                    request=fake_request,
                    dependant=dependant,
                    async_exit_stack=stack,
                    embed_body_fields=False,
                    dependency_cache=dependency_cache,
                )
                dep_kwargs = solved_result.values
                if dependency_cache is not None:
                    dependency_cache.update(solved_result.dependency_cache)

            return dep_kwargs, solved_result.errors

        def handle_errors(errors: list[Any] | None) -> None:
            if errors:
                error_details = "\n".join(str(error) for error in errors)
                logger.info(f"Dependency resolution errors: {error_details}")

        def validate_dependant(dependant: Dependant) -> None:
            if dependant.call is None or not callable(dependant.call):
                msg = "The dependant.call attribute must be a callable."
                raise ValueError(msg)

        @wraps(func)
        async def async_call_with_solved_dependencies(*args: Any, **kwargs: Any) -> T:
            dependant = get_dependant(path="command", call=func)
            validate_dependant(dependant)
            deps, errors = await resolve_dependencies(dependant)
            handle_errors(errors)

            return await cast(Callable[..., Coroutine[Any, Any, T]], dependant.call)(
                *args, **{**deps, **kwargs}
            )

        @wraps(func)
        def sync_call_with_solved_dependencies(*args: Any, **kwargs: Any) -> T:
            dependant = get_dependant(path="command", call=func)
            validate_dependant(dependant)
            deps, errors = asyncio.run(resolve_dependencies(dependant))
            handle_errors(errors)

            return cast(Callable[..., T], dependant.call)(*args, **{**deps, **kwargs})

        return (
            async_call_with_solved_dependencies
            if is_async
            else sync_call_with_solved_dependencies
        )

    if func is None:
        return _impl  # type: ignore
    return _impl(func)