https://linen.dev logo
Join Slack
Powered by
Is there a way to have a per-stream variable added...
# singer-tap-development
r
Is there a way to have a per-stream variable added when running a tap? For example, if there is an endpoint called 
/users
and it takes query parameters (eg: ?group={name}) and other endpoints that also take query parameters (eg 
/licenses?app={name}
), could a user of the tap specify a different query for each endpoint as a stream config? Or would I need to create a tap parameter for each stream so a user can pass it along in the query parameters?
a
Hi, @robby_robinson! To confirm/clarify to make sure I understand: do you mean that as a tap developer, you want to be able to accept per-stream config overrides? And for the use case above, to allow those to be passed to the endpoint when provided by the user?
r
Yes, each stream would be able to have custom query parameters applied and defined by the tap user
a
Thanks for confirming. The SDK doesn't implement anything like this for you automatically as of now, and I don't know of any prior implementations to reference, but it can definitely be done. What I'd probably recommend is to add a tap config option called "stream_overrides" or "stream_config" or similar, defined as an ObjectType, which translates to a Python dictionary. The tap config would look something like this:
Copy code
{
  // Other tap-wide settings here ...
  "stream_config": {
    "stream_a": {
      // These only apply to 'stream_a':
      "additional_filters": [
        { "param": "app", "value": "my-app" }
      ]
  }
}
Then, you can either let the tap resolve these config overrides during discovery and send them only to the Stream classes that need them, or (probably easier) just use the default behavior at the tap level (every stream gets the whole config object) and let streams check for overrides that match their own name at runtime, for instance when responding get_url_params(). Does this help?
r
This gives me something to work with; I’ll give it a go!
It looks like this solution will work nicely. Thanks @aaronsteers!
For anyone who may come across this later, this is what the code looks like: # tap.py
Copy code
th.Property(
     'stream_config',
     th.ArrayType(
         th.PropertiesList(
             th.Property(
                 'stream',
                 th.StringType,
                 description='Name of stream to apply config.'
             ),
             th.Property(
                 'params',
                 th.ArrayType(
                     th.PropertiesList(
                         th.Property(
                             'param',
                             th.StringType,
                             description='Name of query parameter.'
                         ),
                         th.Property(
                             'value',
                             th.StringType,
                             description='Value of query parameter.'
                         )
                     )
                 )
             )
         )
     ),
     description='Custom config for stream.'
# client.py
Copy code
def get_url_params(
        self, context: Optional[dict], next_page_token: Optional[Any]
    ) -> Dict[str, Any]:
        stream_config = self.config.get('stream_config')
        
        if stream_config:
            params = [c.get('params') for c in stream_config if c.get('stream') == self.name]
            
            if params:
                return {p.get('param'):p.get('value') for p in params[-1]}
and the config.json
Copy code
"stream_config": [
        {
            "stream": "users",
            "params": [
                {"param": "$count", "value": "true"},
                {"param": "$select", "value": "id,mail"}
            ]
        }
    ]
a
thank you2
Open in Slack
PreviousNext
Powered by