Feat: dlt.destination as named destination factory (#3204)

* dlt.destination as factory initializer, existing test adjusted

* Additional test adjustments

* test fix in test_duckdb_client.py

* Test readability improvements

* Main destination function in decorators.py improved

* Docs improvements with separate named destination section

* Undo unnecessary adjustments in tests

* custom destination test added

* fix in destination snippet example

* Fix broken links in docs

Author: anuunchin

dlt/destinations/decorators.py

@@ -1,9 +1,7 @@
-import functools
-
 import inspect
 from typing import Any, Type, Optional, Callable, Union, overload
 from typing_extensions import Concatenate
-from dlt.common.destination.reference import DestinationReference
+from dlt.common.destination.reference import DestinationReference, AnyDestination, Destination
 from dlt.common.reflection.spec import get_spec_name_from_f
 from dlt.common.typing import AnyFun
 
@@ -23,28 +21,11 @@
 )
 
 
-@overload
-def destination(
-    func: Callable[
-        Concatenate[Union[TDataItems, str], TTableSchema, TDestinationCallableParams], Any
-    ],
-    /,
-    loader_file_format: TLoaderFileFormat = None,
-    batch_size: int = 10,
-    name: str = None,
-    naming_convention: str = "direct",
-    skip_dlt_columns_and_tables: bool = True,
-    max_table_nesting: int = 0,
-    spec: Type[CustomDestinationClientConfiguration] = None,
-    max_parallel_load_jobs: Optional[int] = None,
-    loader_parallelism_strategy: Optional[TLoaderParallelismStrategy] = None,
-) -> Callable[TDestinationCallableParams, _destination]: ...
-
-
 @overload
 def destination(
     func: None = ...,
     /,
+    *,
     loader_file_format: TLoaderFileFormat = None,
     batch_size: int = 10,
     name: str = None,
@@ -57,12 +38,71 @@ def destination(
 ) -> Callable[
     [Callable[Concatenate[Union[TDataItems, str], TTableSchema, TDestinationCallableParams], Any]],
     Callable[TDestinationCallableParams, _destination],
-]: ...
+]:
+    """A decorator that transforms a function into a custom destination with configuration parameters.
+
+    This overload handles parameterized decorator usage where configuration is provided
+    in the decorator call, and the function to be decorated is applied later.
+
+    #### Example Usage:
+
+    >>> @dlt.destination(batch_size=100, loader_file_format="parquet")
+    >>> def my_destination(items, table, api_url: str = dlt.config.value, api_secret = dlt.secrets.value):
+    >>>     print(table["name"])
+    >>>     print(items)
+    >>>
+    >>> p = dlt.pipeline("chess_pipeline", destination=my_destination)
+
+    Args:
+        func (None): Must be None for this overload. The actual function will be provided
+            when the decorator is applied.
+
+        loader_file_format (TLoaderFileFormat, optional): Defines the format in which files are stored
+            in the load package before being sent to the destination function. Defaults to None.
+
+        batch_size (int, optional): Defines how many items per function call are batched together
+            and sent as an array. If set to 0, instead of passing actual data items, you will receive
+            one call per load job with the path of the file as the "items" argument. Defaults to 10.
+
+        name (str, optional): Defines the name of the destination. If not provided, defaults to
+            the name of the decorated function.
 
+        naming_convention (str, optional): Controls how table and column names are normalized.
+            The default value, "direct", keeps all names unchanged.
 
+        skip_dlt_columns_and_tables (bool, optional): Defines whether internal dlt tables and columns
+            are included in the custom destination function. Defaults to True.
+
+        max_table_nesting (int, optional): Defines how deep the normalizer will go to flatten nested
+            fields in your data to create subtables. This overrides any source settings and defaults
+            to 0, meaning no nested tables are created.
+
+        spec (Type[CustomDestinationClientConfiguration], optional): Defines a configuration spec used
+            to inject arguments into the decorated function. Arguments not included in the spec will
+            not be injected. Defaults to None.
+
+        max_parallel_load_jobs (Optional[int], optional): Defines the maximum number of load jobs
+            that can run concurrently during the load process. Defaults to None.
+
+        loader_parallelism_strategy (Optional[TLoaderParallelismStrategy], optional): Determines the
+            load job parallelism strategy. Can be "sequential" (equivalent to max_parallel_load_jobs=1),
+            "table-sequential" (one load job per table at a time), or "parallel". Defaults to None.
+
+    Returns:
+        Callable[[Callable[Concatenate[Union[TDataItems, str], TTableSchema, TDestinationCallableParams], Any]], Callable[TDestinationCallableParams, _destination]]:
+            A decorator function that accepts the destination function and returns a callable that can
+            be used to create a custom dlt destination instance.
+    """
+    ...
+
+
+@overload
 def destination(
-    func: Optional[AnyFun] = None,
+    func: Callable[
+        Concatenate[Union[TDataItems, str], TTableSchema, TDestinationCallableParams], Any
+    ],
     /,
+    *,
     loader_file_format: TLoaderFileFormat = None,
     batch_size: int = 10,
     name: str = None,
@@ -72,44 +112,114 @@ def destination(
     spec: Type[CustomDestinationClientConfiguration] = None,
     max_parallel_load_jobs: Optional[int] = None,
     loader_parallelism_strategy: Optional[TLoaderParallelismStrategy] = None,
-) -> Any:
-    """A decorator that transforms a function that takes two positional arguments "table" and "items" and any number of keyword arguments with defaults
-    into a callable that will create a custom destination. The function does not return anything, the keyword arguments can be configuration and secrets values.
+) -> Callable[TDestinationCallableParams, _destination]:
+    """Creates a destination factory from a function by directly passing it as an argument.
 
-    #### Example Usage with Configuration and Secrets:
+    This overload handles direct function calls where the destination function is passed
+    as the first argument, rather than using decorator syntax.
+
+    #### Example Usage:
 
-    >>> @dlt.destination(batch_size=100, loader_file_format="parquet")
     >>> def my_destination(items, table, api_url: str = dlt.config.value, api_secret = dlt.secrets.value):
     >>>     print(table["name"])
     >>>     print(items)
     >>>
-    >>> p = dlt.pipeline("chess_pipeline", destination=my_destination)
-
-    Here all incoming data will be sent to the destination function with the items in the requested format and the dlt table schema.
-    The config and secret values will be resolved from the path destination.my_destination.api_url and destination.my_destination.api_secret.
+    >>> dest = dlt.destination(my_destination, batch_size=100, loader_file_format="parquet")
+    >>> p = dlt.pipeline("chess_pipeline", destination=dest)
 
     Args:
-        func (Optional[AnyFun]): A function that takes two positional arguments "table" and "items" and any number of keyword arguments with defaults which will process the incoming data.
+        func (Callable[Concatenate[Union[TDataItems, str], TTableSchema, TDestinationCallableParams], Any]):
+            A callable that takes two positional arguments "items" and "table",
+            followed by any number of keyword arguments with default values. This function
+            will process the incoming data and does not need to return anything. The keyword
+            arguments can represent configuration or secret values.
+
+        loader_file_format (TLoaderFileFormat): Defines the format in which files are stored in the load package
+            before being sent to the destination function.
+
+        batch_size (int): Defines how many items per function call are batched together and sent as an array.
+            If set to 0, instead of passing actual data items, you will receive one call per load job
+            with the path of the file as the "items" argument. You can then open and process that file as needed.
+
+        name (str): Defines the name of the destination created by the destination decorator.
+            Defaults to the name of the function.
 
-        loader_file_format (TLoaderFileFormat): defines in which format files are stored in the load package before being sent to the destination function, this can be puae-jsonl or parquet.
+        naming_convention (str): Controls how table and column names are normalized.
+            The default value, "direct", keeps all names unchanged.
 
-        batch_size (int): defines how many items per function call are batched together and sent as an array. If you set a batch-size of 0, instead of passing in actual dataitems, you will receive one call per load job with the path of the file as the items argument. You can then open and process that file in any way you like.
+        skip_dlt_columns_and_tables (bool): Defines whether internal dlt tables and columns
+            are included in the custom destination function. Defaults to True.
 
-        name (str): defines the name of the destination that gets created by the destination decorator, defaults to the name of the function
+        max_table_nesting (int): Defines how deep the normalizer will go to flatten nested fields
+            in your data to create subtables. This overrides any source settings and defaults to 0,
+            meaning no nested tables are created.
 
-        naming_convention (str): defines the name of the destination that gets created by the destination decorator. This controls how table and column names are normalized. The default is direct which will keep all names the same.
+        spec (Type[CustomDestinationClientConfiguration]): Defines a configuration spec used
+            to inject arguments into the decorated function. Arguments not included in the spec will not be injected.
+
+        max_parallel_load_jobs (Optional[int]): Defines the maximum number of load jobs
+            that can run concurrently during the load process.
+
+        loader_parallelism_strategy (Optional[TLoaderParallelismStrategy]): Determines the load job parallelism strategy.
+            Can be "sequential" (equivalent to max_parallel_load_jobs=1), "table-sequential" (one load job per table at a time),
+            or "parallel".
+
+    Returns:
+        Callable[TDestinationCallableParams, _destination]: A callable that can be used to create a custom dlt destination instance.
+    """
+    ...
+
+
+@overload
+def destination(
+    destination_name: str,
+    /,
+    *,
+    destination_type: Optional[str] = None,
+    credentials: Optional[Any] = None,
+    **kwargs: Any,
+) -> AnyDestination:
+    """Instantiates a destination from the provided destination name and type by retrieving the corresponding
+    destination factory and initializing it with the given credentials and keyword arguments.
 
-        skip_dlt_columns_and_tables (bool): defines wether internal tables and columns will be fed into the custom destination function. This is set to True by default.
+    Args:
+        destination_name (str): The name of the destination instance to initialize.
 
-        max_table_nesting (int): defines how deep the normalizer will go to normalize nested fields on your data to create subtables. This overwrites any settings on your source and is set to zero to not create any nested tables by default.
+        destination_type (Optional[str]): The type of the destination to instantiate.
 
-        spec (Type[CustomDestinationClientConfiguration]): defines a configuration spec that will be used to to inject arguments into the decorated functions. Argument not in spec will not be injected
+        credentials (Optional[Any]): Credentials used to connect to the destination.
+            May be an instance of a credential class supported by the respective destination.
 
-        max_parallel_load_jobs (Optional[int]): how many load jobs at most will be running during the load
+        **kwargs (Any): Additional keyword arguments passed to the destination factory.
 
-        loader_parallelism_strategy (Optional[TLoaderParallelismStrategy]): Can be "sequential" which equals max_parallel_load_jobs=1, "table-sequential" where each table will have at most one loadjob at any given time and "parallel"
     Returns:
-        Any: A callable that can be used to create a dlt custom destination instance
+        AnyDestination: An initialized destination.
+    """
+    ...
+
+
+def destination(
+    func_or_name: Union[Optional[AnyFun], str] = None,
+    /,
+    *,
+    loader_file_format: TLoaderFileFormat = None,
+    batch_size: int = 10,
+    name: str = None,
+    naming_convention: str = "direct",
+    skip_dlt_columns_and_tables: bool = True,
+    max_table_nesting: int = 0,
+    spec: Type[CustomDestinationClientConfiguration] = None,
+    max_parallel_load_jobs: Optional[int] = None,
+    loader_parallelism_strategy: Optional[TLoaderParallelismStrategy] = None,
+    **kwargs: Any,
+) -> Any:
+    """When used as a decorator, transforms a function that takes two positional arguments, "items" and "table",
+    along with any number of keyword arguments with default values, into a callable that will create a custom destination.
+    The function itself does not return anything, the keyword arguments can represent configuration or secret values.
+
+    When used as a function with the first argument being a string (the destination name),
+    instantiates a destination from the provided destination name and type by retrieving the corresponding
+    destination factory and initializing it with the given credentials and keyword arguments.
     """
 
     def decorator(
@@ -165,9 +275,31 @@ def wrapper(
         setattr(wrapper, "_factory", D)  # noqa
         return wrapper
 
-    if func is None:
+    if func_or_name is None:
         # we're called with parens.
         return decorator
+    elif not isinstance(func_or_name, str):
+        # we're called as @dlt.destination without parens.
+        return decorator(func_or_name)
+
+    # Factory mode: create destination instance from given string
+    destination_type = kwargs.pop("destination_type", None)
+    if destination_type:
+        destination = Destination.from_reference(
+            ref=destination_type,
+            destination_name=func_or_name,
+            **kwargs,
+        )
+    elif kwargs.get("destination_callable"):
+        destination = Destination.from_reference(
+            ref="destination",
+            destination_name=func_or_name,
+            **kwargs,
+        )
+    else:
+        destination = Destination.from_reference(
+            ref=func_or_name,
+            **kwargs,
+        )
 
-    # we're called as @dlt.destination without parens.
-    return decorator(func)
+    return destination

docs/website/docs/dlt-ecosystem/destinations/destination.md

@@ -129,35 +129,31 @@ There are multiple ways to pass the custom destination function to the `dlt` pip
   p = dlt.pipeline("my_pipe", destination=my_destination(api_key=os.getenv("API_KEY"))) # type: ignore[call-arg]
   ```
 
-- Directly via destination reference. In this case, don't use the decorator for the destination function.
+- Via the `dlt.destination()` function that initializes the destination. In this case, don't use the decorator for the destination function.
   ```py
   # File my_destination.py
 
-  from dlt.common.destination import Destination
-
   # Don't use the decorator
   def local_destination_func(items: TDataItems, table: TTableSchema) -> None:
       ...
 
-  # Via destination reference
+  # Via dlt.destination() that initializes the destination
   p = dlt.pipeline(
       "my_pipe",
-      destination=Destination.from_reference(
-          "destination", destination_callable=local_destination_func
+      destination=dlt.destination(
+          "my_destination", destination_callable=local_destination_func
       )
   )
   ```
 - Via a fully qualified string to function location (this can be set in `config.toml` or through environment variables). The destination function should be located in another file.
   ```py
   # File my_pipeline.py
 
-  from dlt.common.destination import Destination
-
   # Fully qualified string to function location
   p = dlt.pipeline(
       "my_pipe",
-      destination=Destination.from_reference(
-          "destination", destination_callable="my_destination.local_destination_func"
+      destination=dlt.destination(
+          "my_destination", destination_callable="my_destination.local_destination_func"
       )
   )
   ```

docs/website/docs/general-usage/credentials/setup.md

@@ -747,7 +747,7 @@ You have additional options for using multiple instances of the same source:
 
 1. Use the `clone()` method as explained in the [sql_database documentation](../../dlt-ecosystem/verified-sources/sql_database/advanced.md#configure-many-sources-side-by-side-with-custom-sections).
 
-2. Create [named destinations](../destination.md#configure-multiple-destinations-in-a-pipeline) to use the same destination type with different configurations.
+2. Create [named destinations](../destination.md#configure-multiple-destinations-of-the-same-type) to use the same destination type with different configurations.
 :::
 
 ## Troubleshoot configuration errors