Make a simple pipeline using the scopes declared by the plugin.
(self, scope_config_pairs: list[ScopeConfigPair],
connection: Connection)
| 128 | yield from self.remote_scope_groups(connection) |
| 129 | |
| 130 | def make_pipeline(self, scope_config_pairs: list[ScopeConfigPair], |
| 131 | connection: Connection) -> msg.PipelineData: |
| 132 | """ |
| 133 | Make a simple pipeline using the scopes declared by the plugin. |
| 134 | """ |
| 135 | plan = self.make_pipeline_plan(scope_config_pairs, connection) |
| 136 | domain_scopes = [] |
| 137 | for tool_scope, _ in scope_config_pairs: |
| 138 | for scope in self.domain_scopes(tool_scope): |
| 139 | scope.id = tool_scope.domain_id() |
| 140 | scope.raw_data_params = raw_data_params(connection.id, tool_scope.id) |
| 141 | scope.raw_data_table = self._raw_scope_table_name() |
| 142 | domain_scopes.append( |
| 143 | msg.DynamicDomainScope( |
| 144 | type_name=type(scope).__name__, |
| 145 | data=scope.json(exclude_unset=True, by_alias=True) |
| 146 | ) |
| 147 | ) |
| 148 | return msg.PipelineData( |
| 149 | plan=plan, |
| 150 | scopes=domain_scopes |
| 151 | ) |
| 152 | |
| 153 | def make_pipeline_plan(self, scope_config_pairs: list[ScopeConfigPair], |
| 154 | connection: Connection) -> list[list[msg.PipelineTask]]: |
nothing calls this directly
no test coverage detected