Skip to content
Open
Show file tree
Hide file tree
Changes from 9 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
2 changes: 1 addition & 1 deletion src/firebase_functions/params.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ def _quote_if_string(literal: _T) -> _T:
return _obj_cel_name(literal) if not isinstance(literal, str) else f'"{literal}"'


_params: dict[str, Expression] = {}
_params: dict[str, "Param[_typing.Any] | SecretParam"] = {}


@_dataclasses.dataclass(frozen=True)
Expand Down
70 changes: 45 additions & 25 deletions src/firebase_functions/private/manifest.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,26 @@

import dataclasses as _dataclasses
import typing as _typing
from collections.abc import Mapping as _Mapping
from collections.abc import Sequence as _Sequence
from enum import Enum as _Enum
from zoneinfo import ZoneInfo as _ZoneInfo

import typing_extensions as _typing_extensions

import firebase_functions.params as _params
import firebase_functions.private.util as _util

ManifestParamBase = _params.Param | _params.SecretParam

SpecValue: _typing.TypeAlias = (
str | int | float | bool | _util.Sentinel | list["SpecValue"] | dict[str, "SpecValue"] | None
)


class _DataclassInstance(_typing.Protocol):
__dataclass_fields__: _typing.ClassVar[dict[str, _dataclasses.Field[object]]]


class SecretEnvironmentVariable(_typing.TypedDict):
key: _typing_extensions.Required[str]
Expand Down Expand Up @@ -148,7 +161,7 @@ class ManifestEndpoint:
"""A definition of a function as appears in the Manifest."""

entryPoint: str | None = None
region: list[str] | None = _dataclasses.field(default_factory=list[str])
region: list[str] | None = _dataclasses.field(default_factory=list)
platform: str | None = "gcfv2"
availableMemoryMb: int | _params.Expression[int] | _util.Sentinel | None = None
maxInstances: int | _params.Expression[int] | _util.Sentinel | None = None
Expand All @@ -161,7 +174,7 @@ class ManifestEndpoint:
labels: dict[str, str] | None = None
ingressSettings: str | None | _util.Sentinel = None
secretEnvironmentVariables: list[SecretEnvironmentVariable] | _util.Sentinel | None = (
_dataclasses.field(default_factory=list[SecretEnvironmentVariable])
_dataclasses.field(default_factory=list)
)
httpsTrigger: HttpsTrigger | None = None
callableTrigger: CallableTrigger | None = None
Expand All @@ -180,18 +193,16 @@ class ManifestRequiredApi(_typing.TypedDict):
class ManifestStack:
endpoints: dict[str, ManifestEndpoint]
specVersion: str = "v1alpha1"
params: list[_typing.Any] | None = _dataclasses.field(default_factory=list[_typing.Any])
requiredAPIs: list[ManifestRequiredApi] = _dataclasses.field(
default_factory=list[ManifestRequiredApi]
)
params: _Sequence[ManifestParamBase] | None = _dataclasses.field(default_factory=list)
requiredAPIs: list[ManifestRequiredApi] = _dataclasses.field(default_factory=list)


def _param_input_to_spec(
param_input: _params.TextInput
| _params.ResourceInput
| _params.SelectInput
| _params.MultiSelectInput,
) -> dict[str, _typing.Any]:
) -> dict[str, SpecValue]:
if isinstance(param_input, _params.TextInput):
return {
"text": {
Expand Down Expand Up @@ -233,8 +244,8 @@ def _param_input_to_spec(
return {}


def _param_to_spec(param: _params.Param | _params.SecretParam) -> dict[str, _typing.Any]:
spec_dict: dict[str, _typing.Any] = {
def _param_to_spec(param: ManifestParamBase) -> dict[str, SpecValue]:
spec_dict: dict[str, SpecValue] = {
"name": param.name,
"label": param.label,
"description": param.description,
Expand Down Expand Up @@ -266,45 +277,54 @@ def _param_to_spec(param: _params.Param | _params.SecretParam) -> dict[str, _typ
return _dict_to_spec(spec_dict)


def _object_to_spec(data) -> object:
def _object_to_spec(data: object) -> SpecValue:
if isinstance(data, _Enum):
return data.value
result: SpecValue = data.value
elif isinstance(data, _params.Expression):
return f"{data}"
result = f"{data}"
elif isinstance(data, _ZoneInfo):
result = data.key
elif _dataclasses.is_dataclass(data):
return _dataclass_to_spec(data)
elif isinstance(data, list):
return list(map(_object_to_spec, data))
elif isinstance(data, dict):
return _dict_to_spec(data)
result = _dataclass_to_spec(_typing.cast(_DataclassInstance, data))
elif isinstance(data, _Mapping):
result = _dict_to_spec(data)
elif isinstance(data, _Sequence) and not isinstance(data, str | bytes | bytearray):
result = list(map(_object_to_spec, data))
elif data is None:
result = None
elif isinstance(data, _util.Sentinel):
result = data
elif isinstance(data, str | int | float | bool):
result = data
else:
return data
raise TypeError(f"Unsupported manifest spec value: {type(data)!r}")
Comment thread
IzaakGough marked this conversation as resolved.
return result


def _dict_factory(data: list[tuple[str, _typing.Any]]) -> dict:
out: dict = {}
def _dict_factory(data: list[tuple[str, object]]) -> dict[str, SpecValue]:
out: dict[str, SpecValue] = {}
for key, value in data:
if value is not None:
out[key] = _object_to_spec(value)
return out


def _dataclass_to_spec(data) -> dict:
out: dict = {}
def _dataclass_to_spec(data: _DataclassInstance) -> dict[str, SpecValue]:
out: dict[str, SpecValue] = {}
for field in _dataclasses.fields(data):
value = _object_to_spec(getattr(data, field.name))
if value is not None:
out[field.name] = value
return out


def _dict_to_spec(data: dict) -> dict:
def _dict_to_spec(data: _Mapping[str, object]) -> dict[str, SpecValue]:
return _dict_factory(list(data.items()))


def manifest_to_spec_dict(manifest: ManifestStack) -> dict:
def manifest_to_spec_dict(manifest: ManifestStack) -> dict[str, SpecValue]:
params = manifest.params
out: dict = _dataclass_to_spec(manifest)
out: dict[str, SpecValue] = _dataclass_to_spec(manifest)
if params is not None:
out["params"] = list(map(_param_to_spec, params))
return out
Loading