add mark function to emit resource hints from decorated function

dlt/common/source.py

@@ -34,7 +34,10 @@ def unset_current_pipe_name() -> None:
 
 
 def get_current_pipe_name() -> str:
-    """Gets pipe name associated with current thread"""
+    """When executed from withing dlt.resource decorated function, gets pipe name associated with current thread.
+
+    Pipe name is the same as resource name for all currently known cases. In some multithreading cases, pipe name may be not available.
+    """
     name = _CURRENT_PIPE_NAME.get(threading.get_ident())
     if name is None:
         raise ResourceNameNotAvailable()

dlt/extract/__init__.py

@@ -1,4 +1,5 @@
-from dlt.extract.resource import DltResource, with_table_name
+from dlt.extract.resource import DltResource, with_table_name, with_hints
+from dlt.extract.hints import make_hints
 from dlt.extract.source import DltSource
 from dlt.extract.decorators import source, resource, transformer, defer
 from dlt.extract.incremental import Incremental
@@ -8,6 +9,8 @@
     "DltResource",
     "DltSource",
     "with_table_name",
+    "with_hints",
+    "make_hints",
     "source",
     "resource",
     "transformer",

dlt/extract/decorators.py

@@ -37,6 +37,7 @@
     TSchemaContract,
     TTableFormat,
 )
+from dlt.extract.hints import make_hints
 from dlt.extract.utils import (
     ensure_table_schema_columns_hint,
     simulate_func_call,
@@ -48,6 +49,7 @@
 from dlt.common.typing import AnyFun, ParamSpec, Concatenate, TDataItem, TDataItems
 from dlt.common.utils import get_callable_name, get_module_name, is_inner_callable
 from dlt.extract.exceptions import (
+    CurrentSourceNotAvailable,
     DynamicNameNotStandaloneResource,
     InvalidTransformerDataTypeGeneratorFunctionRequired,
     ResourceFunctionExpected,
@@ -56,7 +58,7 @@
     SourceIsAClassTypeError,
     ExplicitSourceNameInvalid,
     SourceNotAFunction,
-    SourceSchemaNotAvailable,
+    CurrentSourceSchemaNotAvailable,
 )
 from dlt.extract.incremental import IncrementalResourceWrapper
 
@@ -67,7 +69,7 @@
 
 @configspec
 class SourceSchemaInjectableContext(ContainerInjectableContext):
-    """A context containing the source schema, present when decorated function is executed"""
+    """A context containing the source schema, present when dlt.source/resource decorated function is executed"""
 
     schema: Schema
 
@@ -78,6 +80,19 @@ class SourceSchemaInjectableContext(ContainerInjectableContext):
         def __init__(self, schema: Schema = None) -> None: ...
 
 
+@configspec
+class SourceInjectableContext(ContainerInjectableContext):
+    """A context containing the source schema, present when dlt.resource decorated function is executed"""
+
+    source: DltSource
+
+    can_create_default: ClassVar[bool] = False
+
+    if TYPE_CHECKING:
+
+        def __init__(self, source: DltSource = None) -> None: ...
+
+
 TSourceFunParams = ParamSpec("TSourceFunParams")
 TResourceFunParams = ParamSpec("TResourceFunParams")
 TDltSourceImpl = TypeVar("TDltSourceImpl", bound=DltSource, default=DltSource)
@@ -395,7 +410,7 @@ def resource(
     def make_resource(
         _name: str, _section: str, _data: Any, incremental: IncrementalResourceWrapper = None
     ) -> DltResource:
-        table_template = DltResource.new_table_template(
+        table_template = make_hints(
             table_name,
             write_disposition=write_disposition,
             columns=columns,
@@ -694,7 +709,15 @@ def get_source_schema() -> Schema:
     try:
         return Container()[SourceSchemaInjectableContext].schema
     except ContextDefaultCannotBeCreated:
-        raise SourceSchemaNotAvailable()
+        raise CurrentSourceSchemaNotAvailable()
+
+
+def get_source() -> DltSource:
+    """When executed from the function decorated with @dlt.resource, returns currently extracted source"""
+    try:
+        return Container()[SourceInjectableContext].source
+    except ContextDefaultCannotBeCreated:
+        raise CurrentSourceNotAvailable()
 
 
 TBoundItems = TypeVar("TBoundItems", bound=TDataItems)

dlt/extract/exceptions.py

@@ -377,14 +377,22 @@ def __init__(self, source_name: str, _typ: Type[Any]) -> None:
         )
 
 
-class SourceSchemaNotAvailable(DltSourceException):
+class CurrentSourceSchemaNotAvailable(DltSourceException):
     def __init__(self) -> None:
         super().__init__(
             "Current source schema is available only when called from a function decorated with"
             " dlt.source or dlt.resource"
         )
 
 
+class CurrentSourceNotAvailable(DltSourceException):
+    def __init__(self) -> None:
+        super().__init__(
+            "Current source is available only when called from a function decorated with"
+            " dlt.resource or dlt.transformer during the extract step"
+        )
+
+
 class ExplicitSourceNameInvalid(DltSourceException):
     def __init__(self, source_name: str, schema_name: str) -> None:
         self.source_name = source_name

dlt/extract/extract.py

@@ -31,7 +31,7 @@
 from dlt.common.storages.load_package import ParsedLoadJobFileName
 from dlt.common.utils import get_callable_name, get_full_class_name
 
-from dlt.extract.decorators import SourceSchemaInjectableContext
+from dlt.extract.decorators import SourceInjectableContext, SourceSchemaInjectableContext
 from dlt.extract.exceptions import DataItemRequiredForDynamicTableHints
 from dlt.extract.pipe import PipeIterator
 from dlt.extract.source import DltSource
@@ -322,7 +322,9 @@ def extract(
     ) -> str:
         # generate load package to be able to commit all the sources together later
         load_id = self.extract_storage.create_load_package(source.discover_schema())
-        with Container().injectable_context(SourceSchemaInjectableContext(source.schema)):
+        with Container().injectable_context(
+            SourceSchemaInjectableContext(source.schema)
+        ), Container().injectable_context(SourceInjectableContext(source)):
             # inject the config section with the current source name
             with inject_section(
                 ConfigSectionContext(

dlt/extract/extractors.py

@@ -18,7 +18,7 @@
     TTableSchemaColumns,
     TPartialTableSchema,
 )
-
+from dlt.extract.hints import HintsMeta
 from dlt.extract.resource import DltResource
 from dlt.extract.typing import TableNameMeta
 from dlt.extract.storage import ExtractStorage, ExtractorItemStorage
@@ -85,6 +85,12 @@ def item_format(items: TDataItems) -> Optional[TLoaderFileFormat]:
 
     def write_items(self, resource: DltResource, items: TDataItems, meta: Any) -> None:
         """Write `items` to `resource` optionally computing table schemas and revalidating/filtering data"""
+        if isinstance(meta, HintsMeta):
+            # update the resource with new hints, remove all caches so schema is recomputed
+            # and contracts re-applied
+            resource.merge_hints(meta.hints)
+            self._reset_contracts_cache()
+
         if table_name := self._get_static_table_name(resource, meta):
             # write item belonging to table with static name
             self._write_to_static_table(resource, table_name, items)
@@ -152,7 +158,7 @@ def _compute_and_update_table(
         self, resource: DltResource, table_name: str, items: TDataItems
     ) -> TDataItems:
         """
-        Computes new table and does contract checks, if false is returned, the table may not be created and not items should be written
+        Computes new table and does contract checks, if false is returned, the table may not be created and no items should be written
         """
         computed_table = self._compute_table(resource, items)
         # overwrite table name (if coming from meta)
@@ -190,6 +196,12 @@ def _compute_and_update_table(
                     filtered_columns[name] = mode
         return items
 
+    def _reset_contracts_cache(self) -> None:
+        """Removes all cached contracts, filtered columns and tables"""
+        self._table_contracts.clear()
+        self._filtered_tables.clear()
+        self._filtered_columns.clear()
+
 
 class JsonLExtractor(Extractor):
     file_format = "puae-jsonl"