Skip to content

Commit

Permalink
Fix mypy errors
Browse files Browse the repository at this point in the history
  • Loading branch information
sultaniman committed Apr 2, 2024
1 parent c28683c commit d344158
Show file tree
Hide file tree
Showing 2 changed files with 34 additions and 33 deletions.
41 changes: 12 additions & 29 deletions dlt/destinations/impl/filesystem/configuration.py
Original file line number Diff line number Diff line change
@@ -1,23 +1,20 @@
import dataclasses

from typing import Any, Callable, Dict, Final, Type, Optional, TypeAlias, Union
from typing import Any, Callable, Dict, Final, Type, Optional, Union


from pendulum.datetime import DateTime
from dlt.common.configuration import configspec, resolve_type
from dlt.common.destination.reference import (
CredentialsConfiguration,
DestinationClientStagingConfiguration,
)
from dlt.common.storages import FilesystemConfiguration
from pendulum.datetime import DateTime
from typing_extensions import TypeAlias

TCurrentDatetimeCallback: TypeAlias = Callable[[], DateTime]
"""A callback which should return current datetime"""

TDatetimeFormatterCallback: TypeAlias = Callable[[DateTime], str]
"""A callback which is responsible to format datetime"""

TDatetimeFormat: TypeAlias = Union[str, TDatetimeFormatterCallback]
TDatetimeFormat: TypeAlias = str
"""Datetime format or formatter callback"""

TLayoutParamCallback: TypeAlias = Callable[[str, str, str, str, str, DateTime], str]
Expand All @@ -27,12 +24,12 @@


@configspec
class FilesystemDestinationClientConfiguration(
class FilesystemDestinationClientConfiguration( # type: ignore[misc]
FilesystemConfiguration, DestinationClientStagingConfiguration
): # type: ignore[misc]
destination_type: Final[str] = dataclasses.field(
):
destination_type: Final[str] = dataclasses.field( # type: ignore
default="filesystem", init=False, repr=False, compare=False
) # type: ignore
)
current_datetime: Optional[Union[DateTime, TCurrentDatetimeCallback]] = None
datetime_format: Optional[TDatetimeFormat] = None
extra_params: Optional[Dict[str, Union[str, TLayoutParamCallback]]] = None
Expand All @@ -41,8 +38,8 @@ class FilesystemDestinationClientConfiguration(
def resolve_credentials_type(self) -> Type[CredentialsConfiguration]:
# use known credentials or empty credentials for unknown protocol
return (
self.PROTOCOL_CREDENTIALS.get(self.protocol) or Optional[CredentialsConfiguration]
) # type: ignore[return-value]
self.PROTOCOL_CREDENTIALS.get(self.protocol) or Optional[CredentialsConfiguration] # type: ignore[return-value]
)

def on_resolved(self) -> None:
"""Resolve configuration for filesystem destination
Expand All @@ -56,22 +53,8 @@ def on_resolved(self) -> None:
* datetime_format,
* extra_params
"""
# If current_datetime is a callable
# then we need to inspect it's return type
# if return type is not DateTime
# then call it and check it's instance
# if it is not DateTime then exit.
current_datetime = self.kwargs.get("current_datetime", self.current_datetime)
if current_datetime is not None:
if callable(current_datetime):
result = current_datetime()
if isinstance(result, DateTime):
self.current_datetime = result
else:
raise RuntimeError(
"current_datetime was passed as callable but "
"didn't return any instance of pendulum.DateTime"
)
if current_datetime := self.kwargs.get("current_datetime"):
self.current_datetime = current_datetime

datetime_format = self.kwargs.get("datetime_format", self.datetime_format)
if datetime_format is not None:
Expand Down
26 changes: 22 additions & 4 deletions dlt/destinations/impl/filesystem/layout.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import re
from types import TracebackType
from typing import Any, Dict, List, Optional, Self, Sequence, Set, Type
from typing import Any, Dict, List, Optional, Sequence, Set, Type

import pendulum

Expand All @@ -9,6 +9,7 @@
from dlt.destinations.impl.filesystem.configuration import (
FilesystemDestinationClientConfiguration,
)
from typing_extensions import Self

# TODO: ensure layout only has supported placeholders
SUPPORTED_PLACEHOLDERS = {
Expand Down Expand Up @@ -42,7 +43,7 @@ def __init__(
self.job_info = job_info
self.load_id = load_id
self.schema_name = schema_name
self._params = {}
self._params: Dict[str, Any] = {}

self.table_name = None
self.file_id = None
Expand Down Expand Up @@ -85,8 +86,25 @@ def params(self) -> Optional[Dict[str, Any]]:
"ext": self.file_format,
}
)
# If current_datetime is a callable
# then we need to inspect it's return type
# if return type is not DateTime
# then call it and check it's instance
# if it is not DateTime then exit.
if self.config.current_datetime is not None:
if callable(self.config.current_datetime):
result = self.config.current_datetime()
if isinstance(result, pendulum.DateTime):
self.current_datetime = result
else:
raise RuntimeError(
"current_datetime was passed as callable but "
"didn't return any instance of pendulum.DateTime"
)
else:
self.config.current_datetime = pendulum.now()

now = self.config.current_datetime or pendulum.now()
now: pendulum.DateTime = self.config.current_datetime # type: ignore[assignment]
for key, value in self.config.extra_params.items():
if callable(value):
self._params[key] = value(
Expand Down Expand Up @@ -206,7 +224,7 @@ def get_table_prefix_layout(
) as layout:
# fail if table name is not defined
if "table_name" not in layout.placeholders:
raise CantExtractTablePrefix(layout, "{table_name} placeholder not found. ")
raise CantExtractTablePrefix(config.layout, "{table_name} placeholder not found. ")

table_name_index = layout.layout_placeholders.index("table_name")

Expand Down

0 comments on commit d344158

Please sign in to comment.