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,42 @@
 import dataclasses
-from typing import Final, Type, Optional
+
+from typing import 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_placeholders: 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]
+        )

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
@@ -55,3 +57,14 @@ def __init__(
             environment=environment,
             **kwargs,
         )
+
+    def configuration(
+        self, initial_config: FilesystemDestinationClientConfiguration
+    ) -> FilesystemDestinationClientConfiguration:
+        # If current_datetime is not provided then
+        # we need to make sure we have it one per
+        # filesystem destination instance
+        if not initial_config.current_datetime:
+            initial_config.current_datetime = pendulum.now()
+
+        return super().configuration(initial_config)

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,32 +43,14 @@ 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)
         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)}"
@@ -81,7 +68,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 +83,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 +169,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 +208,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