Extend layout placeholder params for filesystem destinations

dlt/common/storages/configuration.py

@@ -1,5 +1,5 @@
 import os
-from typing import TYPE_CHECKING, Any, Literal, Optional, Type, get_args, ClassVar, Dict, Union
+from typing import Any, ClassVar, Dict, Literal, Optional, Type, get_args, Union
 from urllib.parse import urlparse
 
 from dlt.common.configuration import configspec, resolve_type
@@ -13,6 +13,7 @@
     AzureCredentialsWithoutDefaults,
     BaseConfiguration,
 )
+
 from dlt.common.typing import DictStrAny
 from dlt.common.utils import digest128
 
@@ -74,7 +75,7 @@ class FilesystemConfiguration(BaseConfiguration):
         "adl": Union[AzureCredentialsWithoutDefaults, AzureCredentials],
     }
 
-    bucket_url: str = None
+    bucket_url: Optional[str] = None
 
     # should be a union of all possible credentials as found in PROTOCOL_CREDENTIALS
     credentials: FileSystemCredentials = None

dlt/common/storages/file_storage.py

@@ -283,6 +283,10 @@ def from_relative_path_to_wd(self, relative_path: str) -> str:
     def get_file_name_from_file_path(file_path: str) -> str:
         return os.path.basename(file_path)
 
+    @staticmethod
+    def get_dir_name_from_file_path(file_path: str) -> str:
+        return os.path.dirname(file_path)
+
     @staticmethod
     def validate_file_name_component(name: str) -> None:
         # Universal platform bans several characters allowed in POSIX ie. | < \ or "COM1" :)

dlt/destinations/impl/filesystem/configuration.py

@@ -1,19 +1,71 @@
 import dataclasses
-from typing import Final, Type, Optional
+
+from typing import Any, Callable, Dict, Final, Type, Optional, Union
 
 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"""
+
+TDatetimeFormat: TypeAlias = str
+"""Datetime format or formatter callback"""
+
+TLayoutParamCallback: TypeAlias = Callable[[str, str, str, str, str, DateTime], str]
+"""A callback which should return prepared string value the following arguments passed
+`schema name`, `table name`, `load_id`, `file_id`, `extension` and `current_datetime`.
+"""
 
 
 @configspec
-class FilesystemDestinationClientConfiguration(FilesystemConfiguration, DestinationClientStagingConfiguration):  # type: ignore[misc]
-    destination_type: Final[str] = dataclasses.field(default="filesystem", init=False, repr=False, compare=False)  # type: ignore
+class FilesystemDestinationClientConfiguration(  # type: ignore[misc]
+    FilesystemConfiguration, DestinationClientStagingConfiguration
+):
+    destination_type: Final[str] = dataclasses.field(  # type: ignore
+        default="filesystem", init=False, repr=False, compare=False
+    )
+    current_datetime: Optional[Union[DateTime, TCurrentDatetimeCallback]] = None
+    datetime_format: Optional[TDatetimeFormat] = None
+    extra_params: Optional[Dict[str, Union[str, TLayoutParamCallback]]] = None
 
     @resolve_type("credentials")
     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]
+        return (
+            self.PROTOCOL_CREDENTIALS.get(self.protocol) or Optional[CredentialsConfiguration]  # type: ignore[return-value]
+        )
+
+    def on_resolved(self) -> None:
+        """Resolve configuration for filesystem destination
+
+        The following three variables will override the ones from configuration
+        if supplied via `filesystem(...)` constructor, additionally
+        when provided `extra_params` will be merged with the values from configuration
+        however provided values will always override config.toml.
+
+            * current_datetime,
+            * datetime_format,
+            * extra_params
+        """
+        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:
+            self.datetime_format = datetime_format
+
+        extra_params_arg: Dict[str, Any] = self.kwargs.get("extra_params", {})
+        if not self.extra_params and extra_params_arg:
+            self.extra_params = {}
+
+        if extra_params_arg:
+            for key, val in extra_params_arg.items():
+                self.extra_params[key] = val
+
+        super().on_resolved()

dlt/destinations/impl/filesystem/factory.py

@@ -1,5 +1,7 @@
 import typing as t
 
+import pendulum
+
 from dlt.destinations.impl.filesystem.configuration import FilesystemDestinationClientConfiguration
 from dlt.destinations.impl.filesystem import capabilities
 from dlt.common.destination import Destination, DestinationCapabilitiesContext
@@ -27,6 +29,9 @@ def __init__(
         credentials: t.Union[FileSystemCredentials, t.Dict[str, t.Any], t.Any] = None,
         destination_name: t.Optional[str] = None,
         environment: t.Optional[str] = None,
+        current_datetime: t.Optional[pendulum.DateTime] = None,
+        datetime_format: t.Optional[str] = None,
+        extra_params: t.Optional[t.Dict[str, t.Any]] = None,
         **kwargs: t.Any,
     ) -> None:
         """Configure the filesystem destination to use in a pipeline and load data to local or remote filesystem.
@@ -46,12 +51,27 @@ def __init__(
             credentials: Credentials to connect to the filesystem. The type of credentials should correspond to
                 the bucket protocol. For example, for AWS S3, the credentials should be an instance of `AwsCredentials`.
                 A dictionary with the credentials parameters can also be provided.
+            current_datetime: current datetime used instead of datetime generated by dlt.
+            datetime_format: strftime formatting for current_datetime
+            extra_params: custom layout parameters, all unknown parameters will be skipped,
+                values can be primitive types or callables which also should return a primitive type.
             **kwargs: Additional arguments passed to the destination config
         """
+        destinatination_kwargs: t.Dict[str, t.Any] = {}
+        if current_datetime:
+            destinatination_kwargs["current_datetime"] = current_datetime
+
+        if datetime_format:
+            destinatination_kwargs["datetime_format"] = datetime_format
+
+        if extra_params:
+            destinatination_kwargs["extra_params"] = extra_params
+
         super().__init__(
             bucket_url=bucket_url,
             credentials=credentials,
             destination_name=destination_name,
             environment=environment,
+            kwargs=destinatination_kwargs,
             **kwargs,
         )

dlt/destinations/impl/filesystem/filesystem.py

@@ -7,7 +7,7 @@
 
 from dlt.common import logger
 from dlt.common.schema import Schema, TSchemaTables, TTableSchema
-from dlt.common.storages import FileStorage, ParsedLoadJobFileName, fsspec_from_config
+from dlt.common.storages import FileStorage, fsspec_from_config
 from dlt.common.destination import DestinationCapabilitiesContext
 from dlt.common.destination.reference import (
     NewLoadJob,
@@ -20,9 +20,14 @@
 
 from dlt.destinations.job_impl import EmptyLoadJob
 from dlt.destinations.impl.filesystem import capabilities
-from dlt.destinations.impl.filesystem.configuration import FilesystemDestinationClientConfiguration
+from dlt.destinations.impl.filesystem.configuration import (
+    FilesystemDestinationClientConfiguration,
+)
 from dlt.destinations.job_impl import NewReferenceJob
-from dlt.destinations import path_utils
+from dlt.destinations.impl.filesystem.layout import (
+    get_table_prefix_layout,
+    make_filename,
+)
 
 
 class LoadFilesystemJob(LoadJob):
@@ -38,36 +43,28 @@ def __init__(
         file_name = FileStorage.get_file_name_from_file_path(local_path)
         self.config = config
         self.dataset_path = dataset_path
-        self.destination_file_name = LoadFilesystemJob.make_destination_filename(
-            config.layout, file_name, schema_name, load_id
-        )
+        self.destination_file_name = make_filename(config, file_name, schema_name, load_id)
 
         super().__init__(file_name)
         fs_client, _ = fsspec_from_config(config)
-        self.destination_file_name = LoadFilesystemJob.make_destination_filename(
-            config.layout, file_name, schema_name, load_id
-        )
+        self.destination_file_name = make_filename(config, file_name, schema_name, load_id)
+
+        # Make sure directory exists before moving file
+        dir_path = self.make_remote_path(only_dir=True)
+        fs_client.makedirs(dir_path, exist_ok=True)
+
+        # Once directory is created we can move things
         item = self.make_remote_path()
         fs_client.put_file(local_path, item)
 
-    @staticmethod
-    def make_destination_filename(
-        layout: str, file_name: str, schema_name: str, load_id: str
-    ) -> str:
-        job_info = ParsedLoadJobFileName.parse(file_name)
-        return path_utils.create_path(
-            layout,
-            schema_name=schema_name,
-            table_name=job_info.table_name,
-            load_id=load_id,
-            file_id=job_info.file_id,
-            ext=job_info.file_format,
-        )
-
-    def make_remote_path(self) -> str:
-        return (
-            f"{self.config.protocol}://{posixpath.join(self.dataset_path, self.destination_file_name)}"
-        )
+    def make_remote_path(self, only_dir: bool = False) -> str:
+        if only_dir:
+            dir_path = FileStorage.get_dir_name_from_file_path(self.destination_file_name)
+            return f"{self.config.protocol}://{posixpath.join(self.dataset_path, dir_path)}"
+        else:
+            return (
+                f"{self.config.protocol}://{posixpath.join(self.dataset_path, self.destination_file_name)}"
+            )
 
     def state(self) -> TLoadJobState:
         return "completed"
@@ -81,7 +78,9 @@ def create_followup_jobs(self, final_state: TLoadJobState) -> List[NewLoadJob]:
         jobs = super().create_followup_jobs(final_state)
         if final_state == "completed":
             ref_job = NewReferenceJob(
-                file_name=self.file_name(), status="running", remote_path=self.make_remote_path()
+                file_name=self.file_name(),
+                status="running",
+                remote_path=self.make_remote_path(),
             )
             jobs.append(ref_job)
         return jobs
@@ -94,13 +93,17 @@ class FilesystemClient(JobClientBase, WithStagingDataset):
     fs_client: AbstractFileSystem
     fs_path: str
 
-    def __init__(self, schema: Schema, config: FilesystemDestinationClientConfiguration) -> None:
+    def __init__(
+        self,
+        schema: Schema,
+        config: FilesystemDestinationClientConfiguration,
+    ) -> None:
         super().__init__(schema, config)
         self.fs_client, self.fs_path = fsspec_from_config(config)
         self.config: FilesystemDestinationClientConfiguration = config
         # verify files layout. we need {table_name} and only allow {schema_name} before it, otherwise tables
         # cannot be replaced and we cannot initialize folders consistently
-        self.table_prefix_layout = path_utils.get_table_prefix_layout(config.layout)
+        self.table_prefix_layout = get_table_prefix_layout(config)
         self._dataset_path = self.config.normalize_dataset_name(self.schema)
 
     def drop_storage(self) -> None:
@@ -176,6 +179,7 @@ def update_stored_schema(
             self.fs_client.makedirs(directory, exist_ok=True)
         return expected_update
 
+    # FIXME: maybe have to fixup to support all PathParams and create_path updates
     def _get_table_dirs(self, table_names: Iterable[str]) -> Set[str]:
         """Gets unique directories where table data is stored."""
         table_dirs: Set[str] = set()
@@ -214,7 +218,10 @@ def __enter__(self) -> "FilesystemClient":
         return self
 
     def __exit__(
-        self, exc_type: Type[BaseException], exc_val: BaseException, exc_tb: TracebackType
+        self,
+        exc_type: Type[BaseException],
+        exc_val: BaseException,
+        exc_tb: TracebackType,
     ) -> None:
         pass