#3642 feat: Add `Target.create_sink()` factory method Issue created by edgarrmondragon Feature scope Targets (data type handling, batching, SQL object generation, tests, etc.) Description Problem

Target.add_sink()

currently instantiates sinks by calling

sink_class(target=self, ...)

directly. There is no hook for a target subclass to customize how sinks are constructed — for example, to inject a shared resource (a database connection, an auth token, a thread pool) at construction time. The only workaround today is to store a back-reference to the target on the sink manually: class MySink(BatchSink): def __init__(self, target, ...): super().__init__(target, ...) self._my_target: Any = target # type hack — SDK doesn't expose self.target This is awkward because it requires a loosely-typed attribute (

Any

) to avoid a circular import between the sink module and the target module, and it couples the sink's internal implementation to the target's structure. Proposed solution Extract sink construction from

add_sink()

into an overridable

create_sink()

factory method: # Default implementation (backward compatible) def create_sink( self, stream_name: str, schema: dict, key_properties: Sequence[str] | None = None, ) -> Sink: sink_class = self.get_sink_class(stream_name=stream_name) return sink_class( target=self, stream_name=stream_name, schema=schema, key_properties=key_properties, ) # In add_sink — replace the direct instantiation: sink = self.create_sink(stream_name, schema, key_properties) Targets that need to share resources across sinks can then override

create_sink()

and inject them directly via the sink constructor: class TargetADBC(Target): def create_sink(self, stream_name, schema, key_properties=None) -> ADBCSink: return ADBCSink( target=self, stream_name=stream_name, schema=schema, key_properties=key_properties, connection=self.connection, # shared ADBC connection ) This keeps sinks as self-contained objects whose dependencies are explicit in their constructor signature, rather than requiring them to reach back into the target at runtime. Benefits • Explicit over implicit: the sink's dependencies are visible in its constructor signature • No circular import workaround: the sink module never needs to import the target module • Testable in isolation: sinks can be instantiated in tests by passing dependencies directly, without building a full target • Backward compatible: the default

create_sink()

implementation preserves existing behavior exactly Real-world motivation In `target-adbc`, all sinks must share a single ADBC database connection (having two write connections open simultaneously to the same DuckDB file causes silent data loss on commit). Without

create_sink()

, the connection has to be retrieved lazily via a back-reference to the target, requiring

self._adbc_target: Any = target

and a

# type: ignore

comment. meltano/sdk