Add support for Dagster pipelines#27
Open
Resmung0 wants to merge 6 commits into
Open
Conversation
- Implemented DagsterParser using TypeScript-only regex-based extraction. - Detects Software-Defined Assets, Ops, and their dependencies from Python source. - Supports both function-argument dependencies and explicit 'deps' decorator arguments. - Added visual differentiation between assets (artifact nodes) and ops (default nodes). - Integrated Dagster into the Data Processing pipeline category. - Updated webview UI with fallback icons for Dagster framework detection. Co-authored-by: Resmung0 <36093882+Resmung0@users.noreply.github.com>
Contributor
|
👋 Jules, reporting for duty! I'm here to lend a hand with this pull request. When you start a review, I'll add a 👀 emoji to each comment to let you know I've read it. I'll focus on feedback directed at me and will do my best to stay out of conversations between you and other bots or reviewers to keep the noise down. I'll push a commit with your requested changes shortly after. Please note there might be a delay between these steps, but rest assured I'm on the job! For more direct control, you can switch me to Reactive Mode. When this mode is on, I will only act on comments where you specifically mention me with New to Jules? Learn more at jules.google/docs. For security, I will only act on instructions from the user who triggered this task. |
Reviewer's GuideAdds a new TypeScript Dagster parser to the data processing pipeline framework and updates the webview node UI to visually distinguish Dagster assets/ops and show Dagster-specific metadata, plus corresponding tests. Sequence diagram for Dagster file parsing and webview renderingsequenceDiagram
actor User
participant VSCodeExtension
participant DataProcessingPipeline
participant DagsterParser
participant Webview
User->>VSCodeExtension: Open workspace with Dagster Python files
VSCodeExtension->>DataProcessingPipeline: discoverPipelines()
loop For each Python file
DataProcessingPipeline->>DagsterParser: canParse(fileName, content)
DagsterParser-->>DataProcessingPipeline: boolean
alt canParse is true
DataProcessingPipeline->>DagsterParser: parse(content, filePath)
DagsterParser-->>DataProcessingPipeline: PipelineData(nodes, edges)
DataProcessingPipeline->>VSCodeExtension: add Dagster PipelineData
end
end
VSCodeExtension->>Webview: postMessage PipelineData
Webview->>Webview: Map nodes to React Flow elements
Webview->>Webview: Render PipelineNodeItem
Note over Webview: Dagster assets rendered as artifact nodes
Note over Webview: Dagster ops rendered as default nodes
Class diagram for DagsterParser integration into DataProcessingPipelineclassDiagram
class IParser {
<<interface>>
+string name
+canParse(fileName string, content string) boolean
+parse(content string, filePath string) PipelineData
}
class PipelineData {
+string filePath
+string framework
+PipelineNode[] nodes
+PipelineEdge[] edges
}
class PipelineNode {
+string id
+string label
+string type
+any data
}
class PipelineEdge {
+string id
+string source
+string target
}
class DagsterParser {
+string name
+canParse(fileName string, content string) boolean
+parse(content string, filePath string) PipelineData
}
class AirflowParser
class KedroParser
class DVCParser
class DataProcessingPipeline {
+PipelinePatternType type
+IParser[] parsers
}
IParser <|.. DagsterParser
IParser <|.. AirflowParser
IParser <|.. KedroParser
IParser <|.. DVCParser
DataProcessingPipeline "1" o-- "many" IParser
PipelineData "1" o-- "many" PipelineNode
PipelineData "1" o-- "many" PipelineEdge
Flow diagram for DagsterParser.parse decorator and dependency extractionflowchart TD
A_start[Start parse<br/>content, filePath] --> B_init[Initialize nodes, edges, nodeIds]
B_init --> C_scan[Scan content with decoratorRegex<br/>for asset, op, graph_asset, multi_asset]
C_scan -->|match found| D_processMatch[Process decorator match]
C_scan -->|no more matches| K_external[Identify external assets]
D_processMatch --> D1_getId[Extract id from function name]
D1_getId --> D2_checkId{nodeIds has id}
D2_checkId -->|yes| C_scan
D2_checkId -->|no| E_createNode[Create PipelineNode<br/>asset or op]
E_createNode --> E1_addNode[Push node to nodes<br/>add id to nodeIds]
E1_addNode --> F_args[Parse function args<br/>split, trim, drop context and varargs]
F_args --> F1_forArgs[For each arg]
F1_forArgs --> F2_addEdge[Push edge from arg to id]
F2_addEdge --> G_depsArgs[Check decArgs<br/>for deps list]
G_depsArgs -->|deps found| G1_forDep[For each dep]
G_depsArgs -->|no deps| C_scan
G1_forDep --> G2_addDepEdge[Push edge from dep to id]
G2_addDepEdge --> C_scan
K_external --> K1_collectDefined[Collect defined node ids]
K1_collectDefined --> K2_forEdge[For each edge]
K2_forEdge --> K3_missingSource{edge source in definedNodes}
K3_missingSource -->|yes| K2_forEdge
K3_missingSource -->|no| K4_createExternal[Create external PipelineNode<br/>for missing source]
K4_createExternal --> K5_addExternal[Push external node<br/>add id to definedNodes]
K5_addExternal --> K2_forEdge
K2_forEdge -->|done| L_return[Return PipelineData<br/>filePath, framework, nodes, edges]
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 9 issues, and left some high level feedback:
- In
PipelineNodeItem, the artifact branch checksdata.type === 'artifact'instead of the React Flowtypeprop used previously, which can drift from how nodes are actually instantiated; consider consistently using the nodetypefrom React Flow (or ensuringdata.typeis always kept in sync) to avoid mis‑rendering artifacts. - The new
NodeDropdowncomponent types itsiconanditemsprops asany; tightening these to something likeReact.ComponentType<{ size?: number; className?: string }>foriconand a discriminated union keyed byvariantforitemswould make misuse harder and improve editor tooling. - The Dagster parser’s
decoratorRegexanddepsextraction logic assume single-line decorator argument lists and function signatures, which will miss common multi-line Dagster patterns; consider broadening the regex to handle newlines or, if feasible, switching to a more structured parse to better support real-world Dagster code.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `PipelineNodeItem`, the artifact branch checks `data.type === 'artifact'` instead of the React Flow `type` prop used previously, which can drift from how nodes are actually instantiated; consider consistently using the node `type` from React Flow (or ensuring `data.type` is always kept in sync) to avoid mis‑rendering artifacts.
- The new `NodeDropdown` component types its `icon` and `items` props as `any`; tightening these to something like `React.ComponentType<{ size?: number; className?: string }>` for `icon` and a discriminated union keyed by `variant` for `items` would make misuse harder and improve editor tooling.
- The Dagster parser’s `decoratorRegex` and `deps` extraction logic assume single-line decorator argument lists and function signatures, which will miss common multi-line Dagster patterns; consider broadening the regex to handle newlines or, if feasible, switching to a more structured parse to better support real-world Dagster code.
## Individual Comments
### Comment 1
<location path="src/webview/components/PipelineNodeItem.tsx" line_range="7-10" />
<code_context>
- showSweep: false,
- },
-} as const;
+interface NodeDropdownProps {
+ icon: any;
+ label: string;
+ items: any;
+ isExpanded: boolean;
+ onToggle: (e: React.MouseEvent) => void;
</code_context>
<issue_to_address>
**suggestion:** Tighten NodeDropdown prop types instead of using `any` for icon and items.
Using `any` here undermines the type guarantees around `NodeDropdown`, especially since `variant` drives different expectations for `items` (e.g. params object vs deps array with `path`/`snippet`). Consider a discriminated union keyed by `variant` for `items`, and a more specific `icon` type like `React.ComponentType<{ size?: number; className?: string }>` so mismatches are caught at compile time.
Suggested implementation:
```typescript
type NodeDropdownCommonProps = {
icon: React.ComponentType<{ size?: number; className?: string }>;
label: string;
isExpanded: boolean;
onToggle: (e: React.MouseEvent) => void;
};
type NodeDropdownParamsProps = NodeDropdownCommonProps & {
variant: 'params';
items: Record<string, unknown>;
};
type NodeDropdownDepsItem = {
path: string;
snippet?: string;
};
type NodeDropdownDepsProps = NodeDropdownCommonProps & {
variant: 'deps';
items: NodeDropdownDepsItem[];
};
type NodeDropdownProps = NodeDropdownParamsProps | NodeDropdownDepsProps;
```
Depending on the actual shapes used in the rest of the file:
1. Tighten `Record<string, unknown>` in `NodeDropdownParamsProps.items` to the real params object type if it already exists (e.g. reuse a `Params` or similar type from elsewhere).
2. Update any call sites of `NodeDropdown` to ensure:
- When `variant="params"`, `items` is passed as an object matching the params shape.
- When `variant="deps"`, `items` is passed as an array of objects with at least a `path` property and optional `snippet`.
3. If there is an existing type for dependency items (e.g. `Dependency` or similar), replace `NodeDropdownDepsItem` with that type to avoid duplication.
</issue_to_address>
### Comment 2
<location path="src/webview/components/PipelineNodeItem.tsx" line_range="73-82" />
<code_context>
- case 'audio': return <LuMusic4 size={12} />;
- case 'folder': return <Folder size={12} />;
- default: return null;
+ const getStatusConfig = (status?: string) => {
+ switch (status) {
+ case 'running':
+ return {
+ icon: Clock,
+ color: '#3b82f6',
+ animationVariant: { scale: [1, 1.02, 1], transition: { repeat: Infinity, duration: 2 } },
+ showSweep: true,
+ className: 'running',
+ iconColor: '#3b82f6'
+ };
+ case 'success':
+ return { icon: CheckCircle2, color: '#10b981', iconColor: '#10b981', className: 'success' };
+ case 'failed':
+ return { icon: AlertCircle, color: '#ef4444', iconColor: '#ef4444', className: 'failed' };
+ default:
+ return { icon: Circle, color: '#94a3b8', iconColor: '#94a3b8', className: 'idle' };
}
</code_context>
<issue_to_address>
**issue (bug_risk):** The new status config no longer handles the previous `processing` status, which may regress visuals.
The previous `STATUS_CONFIG` treated both `processing` and `running` as sweeping states. `getStatusConfig` now only special-cases `running`, so any `processing` value still emitted upstream will fall back to the idle config and lose the sweep animation. If `processing` remains a valid status elsewhere, either normalize it to `running` before calling `getStatusConfig`, or add an explicit `case 'processing'` that mirrors the `running` config to preserve the visual behavior.
</issue_to_address>
### Comment 3
<location path="src/extension/pipelines/DataProcessingPipeline.ts" line_range="14" />
<code_context>
Object.assign(new AirflowParser(), { patterns: ['**/dags/*.py'] }),
Object.assign(new KedroParser(), { patterns: ['**/src/**/pipeline.py', '**/src/**/pipelines/**/pipeline.py'] }),
Object.assign(new DVCParser(), { patterns: ['**/dvc.yaml', '**/dvc.yml'] }),
+ Object.assign(new DagsterParser(), { patterns: ['**/*.py'] }),
];
}
</code_context>
<issue_to_address>
**suggestion (performance):** Using `**/*.py` for Dagster patterns may be overly broad and impact performance.
This will route every Python file in the repo through `DagsterParser.canParse`, which can be expensive on large codebases. Consider narrowing the glob to known Dagster locations or conventions (e.g. `dagster_*`, `*assets.py`, or specific Dagster project paths) to limit candidates while still covering typical setups.
```suggestion
Object.assign(new DagsterParser(), {
patterns: [
'**/dagster_*/**/*.py', // common Dagster project/package prefix
'**/*_dagster.py', // files explicitly named for Dagster usage
'**/*assets.py', // Dagster assets modules
'**/*jobs.py', // Dagster jobs definitions
'**/*schedules.py', // Dagster schedules
'**/*sensors.py', // Dagster sensors
'**/*repository.py', // Dagster repository definitions
],
}),
```
</issue_to_address>
### Comment 4
<location path="src/extension/parsers/data-processing/DagsterParser.ts" line_range="26-35" />
<code_context>
+ const decoratorRegex = /@(asset|op|graph_asset|multi_asset)(\([^)]*\))?\s+def\s+(\w+)\s*\(([^)]*)\)/g;
</code_context>
<issue_to_address>
**suggestion:** Decorator and argument parsing in DagsterParser is quite brittle and may break on common Python patterns.
This pattern only supports single-line decorators and defs, and a flat argument list. It will break for nested parentheses or commas inside defaults / type hints (e.g. `list[Tuple[str, int]]`, `arg: Dict[str, int] = {"a": 1}`), leading to missing or incorrect edges. If you stick with regex, consider supporting multiline decorators/defs and using an argument splitter that ignores commas inside brackets/parentheses. Otherwise, explicitly limit this to simple signatures and skip complex ones rather than inferring wrong dependencies.
Suggested implementation:
```typescript
}
/**
* Return true if the argument list is "simple" enough to be safely parsed
* with comma-splitting (no nested brackets/parentheses/braces).
*
* Complex signatures are intentionally skipped to avoid inferring incorrect
* dependencies from Dagster decorators.
*/
private isSimpleDagsterSignature(argList: string): boolean {
// If we see any obvious nesting characters, bail out; these are likely to
// contain commas that would confuse a naive splitter.
// We still allow type hints like list[Tuple[str, int]] and simple defaults.
const hasNestedParens =
argList.includes('(') ||
argList.includes(')') ||
argList.includes('{') ||
argList.includes('}');
if (hasNestedParens) {
return false;
}
// Extremely long signatures or ones that contain newlines/complex annotations
// are also treated as complex and skipped.
if (argList.length > 200 || argList.includes('\n')) {
return false;
}
return true;
}
async parse(content: string, filePath: string): Promise<PipelineData> {
```
```typescript
// 1. Extract Assets and Ops from decorators
// Matches @asset, @op, @graph_asset, @multi_asset
// Captures the decorator body and the function definition.
//
// NOTE: This intentionally only supports "simple" function signatures
// without nested parentheses/brackets in the argument list. More complex
// signatures are detected and skipped (see isSimpleDagsterSignature) to
// avoid inferring incorrect dependencies.
const decoratorRegex = /@(asset|op|graph_asset|multi_asset)(\([^)]*\))?\s+def\s+(\w+)\s*\(([^)]*)\)/g;
```
```typescript
let match;
while ((match = decoratorRegex.exec(content)) !== null) {
const [fullMatch, type, decArgs, name, funcArgs] = match;
// Skip complex signatures that we cannot safely parse using simple
// comma-based argument splitting. This avoids emitting incorrect edges.
if (!this.isSimpleDagsterSignature(funcArgs)) {
continue;
}
const id = name;
```
If the argument list from `funcArgs` is further processed elsewhere in this file (for example, split on commas to infer dependencies), that logic will now only run for signatures that pass `isSimpleDagsterSignature`. You may want to tighten or relax the heuristics in `isSimpleDagsterSignature` based on real-world Dagster usage in your codebase (e.g., allow certain benign patterns, or skip more aggressively if you see incorrect edges in practice).
</issue_to_address>
### Comment 5
<location path="tests/DagsterParser.test.ts" line_range="6-8" />
<code_context>
+describe('DagsterParser', () => {
+ const parser = new DagsterParser();
+
+ it('should identify Dagster files', () => {
+ const content = 'from dagster import asset, Definitions\n@asset\ndef my_asset(): pass';
+ expect(parser.canParse('defs.py', content)).toBe(true);
+ });
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add more `canParse` coverage for the different Dagster detection heuristics.
This test only covers the `from dagster import ...` path. Since `canParse` also detects `import dagster`, `@op`, and `definitions(`, please add tests for each of these cases, plus one where none are present, to prevent regressions in the detection logic.
Suggested implementation:
```typescript
import { DagsterParser } from '../src/extension/parsers/data-processing/DagsterParser';
describe('DagsterParser', () => {
const parser = new DagsterParser();
describe('canParse', () => {
it('returns true for files using "from dagster import ..."', () => {
const content = 'from dagster import asset, Definitions\n@asset\ndef my_asset(): pass';
expect(parser.canParse('defs.py', content)).toBe(true);
});
it('returns true for files using "import dagster"', () => {
const content = 'import dagster\n\ndef my_job():\n pass';
expect(parser.canParse('job.py', content)).toBe(true);
});
it('returns true for files using "@op" decorator', () => {
const content = 'from dagster import op\n\n@op\ndef my_op(context):\n pass';
expect(parser.canParse('ops.py', content)).toBe(true);
});
it('returns true for files defining "Definitions("', () => {
const content = 'from dagster import Definitions\n\ndefs = Definitions(assets=[], jobs=[])';
expect(parser.canParse('defs.py', content)).toBe(true);
});
it('returns false for non-Dagster Python files', () => {
const content = 'def not_dagster():\n return "hello"';
expect(parser.canParse('script.py', content)).toBe(false);
});
});
it('should parse assets and their dependencies from arguments', async () => {
const content = `
```
If there are other `describe('DagsterParser'...)` blocks or existing `canParse` tests elsewhere in this file, you may want to consolidate them into this new `describe('canParse', ...)` block for clarity, or adjust the test names to avoid duplication.
</issue_to_address>
### Comment 6
<location path="tests/DagsterParser.test.ts" line_range="11-20" />
<code_context>
+ expect(parser.canParse('defs.py', content)).toBe(true);
+ });
+
+ it('should parse assets and their dependencies from arguments', async () => {
+ const content = `
+from dagster import asset
+
+@asset
+def raw_data():
+ return [1, 2, 3]
+
+@asset
+def processed_data(raw_data):
+ return [x * 10 for x in raw_data]
+`;
+ const result = await parser.parse(content, 'dummy.py');
+ expect(result.framework).toBe('Dagster');
+ expect(result.nodes.length).toBe(2);
+
+ const raw = result.nodes.find(n => n.id === 'raw_data');
+ const processed = result.nodes.find(n => n.id === 'processed_data');
+
+ expect(raw?.type).toBe('artifact');
+ expect(processed?.type).toBe('artifact');
+
+ expect(result.edges.length).toBe(1);
+ expect(result.edges[0]).toEqual({
+ id: 'e-raw_data-processed_data',
+ source: 'raw_data',
</code_context>
<issue_to_address>
**suggestion (testing):** Cover additional argument-related edge cases (no args, *args/**kwargs, positional-only).
Since the parser trims and filters arguments (type hints, `context`, `*` prefix, etc.), add tests for: (1) an asset/op with no arguments (no edges created), (2) functions with `*args`/`**kwargs` to ensure they’re ignored and don’t create nodes, and (3) unusual argument syntax (spacing quirks and positional-only args) to validate the split/trim logic remains robust.
Suggested implementation:
```typescript
it('should parse assets and their dependencies from arguments', async () => {
const content = `
from dagster import asset
@asset
def raw_data():
return [1, 2, 3]
@asset
def processed_data(raw_data):
return [x * 10 for x in raw_data]
`;
const result = await parser.parse(content, 'dummy.py');
expect(result.framework).toBe('Dagster');
expect(result.nodes.length).toBe(2);
const raw = result.nodes.find(n => n.id === 'raw_data');
const processed = result.nodes.find(n => n.id === 'processed_data');
expect(raw?.type).toBe('artifact');
expect(processed?.type).toBe('artifact');
expect(result.edges.length).toBe(1);
expect(result.edges[0]).toEqual({
id: 'e-raw_data-processed_data',
source: 'raw_data',
target: 'processed_data'
});
});
it('should not create edges for assets without arguments', async () => {
const content = `
from dagster import asset
@asset
def upstream_asset():
return [1, 2, 3]
@asset
def downstream_asset():
return "no dependencies here"
`;
const result = await parser.parse(content, 'no_args.py');
expect(result.framework).toBe('Dagster');
// Both assets are discovered as nodes
expect(result.nodes.map(n => n.id).sort()).toEqual(
expect.arrayContaining(['upstream_asset', 'downstream_asset']),
);
// No edges should be created because there are no function arguments
expect(result.edges.length).toBe(0);
});
it('should ignore *args and **kwargs when parsing dependencies', async () => {
const content = `
from dagster import asset
@asset
def flexible_asset(*args, **kwargs):
return sum(args) if args else 0
`;
const result = await parser.parse(content, 'varargs.py');
expect(result.framework).toBe('Dagster');
// Only the asset node should be present
expect(result.nodes.length).toBe(1);
expect(result.nodes[0]?.id).toBe('flexible_asset');
// *args and **kwargs should not create any dependency edges
expect(result.edges.length).toBe(0);
});
it('should handle unusual argument syntax and positional-only args', async () => {
const content = `
from dagster import asset
@asset
def raw_data():
return [1, 2, 3]
@asset
def other_data():
return [4, 5, 6]
@asset
def processed_data( raw_data , other_data , /, *args, context = None, **kwargs ):
return [x * 10 for x in raw_data + other_data]
`;
const result = await parser.parse(content, 'positional_only.py');
expect(result.framework).toBe('Dagster');
// All three assets should be discovered
expect(result.nodes.map(n => n.id).sort()).toEqual(
expect.arrayContaining(['raw_data', 'other_data', 'processed_data']),
);
// Only raw_data and other_data should be treated as dependencies
// The positional-only marker "/", *args, context, and **kwargs must not become nodes/edges
const edgeIds = result.edges.map(e => e.id).sort();
expect(edgeIds).toEqual(
expect.arrayContaining([
'e-raw_data-processed_data',
'e-other_data-processed_data',
]),
);
const sources = result.edges.map(e => e.source).sort();
const targets = [...new Set(result.edges.map(e => e.target))];
expect(sources).toEqual(expect.arrayContaining(['raw_data', 'other_data']));
expect(targets).toEqual(['processed_data']);
});
it('should parse dependencies from deps argument in decorator', async () => {
const content = `
@asset(deps=["upstream_asset"])
def downstream_asset():
pass
`;
```
These tests assume the existing `DagsterParser` already:
1. Ignores arguments named `context`.
2. Strips `*`/`**` prefixes and filters non-identifier tokens (like `/`).
If the parser does not currently implement that behavior, corresponding logic should be added in the argument-parsing portion of `DagsterParser` to:
- Split on commas,
- Trim whitespace,
- Remove type hints and default values (text after `:` or `=`),
- Strip leading `*`/`**`,
- Filter out `context` and any non-identifier tokens before constructing edges.
</issue_to_address>
### Comment 7
<location path="tests/DagsterParser.test.ts" line_range="41-56" />
<code_context>
+ });
+ });
+
+ it('should parse dependencies from deps argument in decorator', async () => {
+ const content = `
+@asset(deps=["upstream_asset"])
+def downstream_asset():
+ pass
+`;
+ const result = await parser.parse(content, 'dummy.py');
+ expect(result.edges.length).toBe(1);
+ expect(result.edges[0]).toEqual({
+ id: 'e-upstream_asset-downstream_asset',
+ source: 'upstream_asset',
+ target: 'downstream_asset'
+ });
+ // Should also detect upstream_asset as an external asset
+ expect(result.nodes.find(n => n.id === 'upstream_asset')).toBeDefined();
+ });
+
</code_context>
<issue_to_address>
**suggestion (testing):** Extend `deps` tests to cover multiple deps and `AssetKey(...)` usage.
The current test only covers a single string dep. To better validate the parsing logic, please also add cases with multiple deps in the list (verifying all edges are created) and with values like `AssetKey("foo")` to confirm `foo` is extracted correctly, matching typical Dagster usage.
```suggestion
it('should parse dependencies from deps argument in decorator', async () => {
const content = `
@asset(deps=["upstream_asset"])
def downstream_asset():
pass
`;
const result = await parser.parse(content, 'dummy.py');
expect(result.edges.length).toBe(1);
expect(result.edges[0]).toEqual({
id: 'e-upstream_asset-downstream_asset',
source: 'upstream_asset',
target: 'downstream_asset',
});
// Should also detect upstream_asset as an external asset
expect(result.nodes.find((n) => n.id === 'upstream_asset')).toBeDefined();
});
it('should parse multiple deps and AssetKey deps in decorator', async () => {
const content = `
from dagster import asset, AssetKey
@asset(deps=["upstream_asset", "another_upstream", AssetKey("keyed_asset")])
def downstream_asset():
pass
`;
const result = await parser.parse(content, 'dummy.py');
expect(result.edges.length).toBe(3);
expect(result.edges).toEqual(
expect.arrayContaining([
{
id: 'e-upstream_asset-downstream_asset',
source: 'upstream_asset',
target: 'downstream_asset',
},
{
id: 'e-another_upstream-downstream_asset',
source: 'another_upstream',
target: 'downstream_asset',
},
{
id: 'e-keyed_asset-downstream_asset',
source: 'keyed_asset',
target: 'downstream_asset',
},
]),
);
// All upstream assets (including AssetKey-based) should be detected as external assets
['upstream_asset', 'another_upstream', 'keyed_asset'].forEach((id) => {
expect(result.nodes.find((n) => n.id === id)).toBeDefined();
});
});
```
</issue_to_address>
### Comment 8
<location path="tests/DagsterParser.test.ts" line_range="58-71" />
<code_context>
+ expect(result.nodes.find(n => n.id === 'upstream_asset')).toBeDefined();
+ });
+
+ it('should handle ops and mixed pipelines', async () => {
+ const content = `
+@asset
+def my_asset(): pass
+
+@op
+def my_op(my_asset): pass
+`;
+ const result = await parser.parse(content, 'dummy.py');
+ const op = result.nodes.find(n => n.id === 'my_op');
+ expect(op?.type).toBe('default');
+ expect(result.edges[0].source).toBe('my_asset');
+ expect(result.edges[0].target).toBe('my_op');
+ });
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add assertions for node metadata (framework and asset/op classification) in mixed pipelines.
To better cover the visualization behavior, please also assert the node metadata for both nodes (e.g. `framework === 'Dagster'`, `data.dataType === 'asset'` for the asset and `'op'` for the op). This helps catch future parser changes that might break the UI integration.
```suggestion
it('should handle ops and mixed pipelines', async () => {
const content = `
@asset
def my_asset(): pass
@op
def my_op(my_asset): pass
`;
const result = await parser.parse(content, 'dummy.py');
const asset = result.nodes.find(n => n.id === 'my_asset');
const op = result.nodes.find(n => n.id === 'my_op');
// Basic node type check
expect(op?.type).toBe('default');
// Framework and classification metadata
expect(asset?.framework).toBe('Dagster');
expect(op?.framework).toBe('Dagster');
expect(asset?.data?.dataType).toBe('asset');
expect(op?.data?.dataType).toBe('op');
// Edge wiring between asset and op
expect(result.edges[0].source).toBe('my_asset');
expect(result.edges[0].target).toBe('my_op');
});
```
</issue_to_address>
### Comment 9
<location path="tests/DagsterParser.test.ts" line_range="73-83" />
<code_context>
+ expect(result.edges[0].target).toBe('my_op');
+ });
+
+ it('should handle type hints in arguments', async () => {
+ const content = `
+@asset
+def typed_asset(upstream: DataFrame, context: AssetExecutionContext):
+ pass
+`;
+ const result = await parser.parse(content, 'dummy.py');
+ expect(result.edges.length).toBe(1);
+ expect(result.edges[0].source).toBe('upstream');
+ });
+});
</code_context>
<issue_to_address>
**suggestion (testing):** Include a test for external-node deduplication when multiple edges reference the same source.
Right now we only cover the single-dependency case. Please also add a test where two downstream assets/ops depend on the same upstream name that isn’t defined locally, and assert that the parser still creates exactly one external node for that shared source. This will verify the `Set`-based deduplication and guard against unexpected graph inflation.
```suggestion
it('should handle type hints in arguments', async () => {
const content = `
@asset
def typed_asset(upstream: DataFrame, context: AssetExecutionContext):
pass
`;
const result = await parser.parse(content, 'dummy.py');
expect(result.edges.length).toBe(1);
expect(result.edges[0].source).toBe('upstream');
});
it('should deduplicate external nodes shared by multiple downstream assets', async () => {
const content = `
from dagster import asset
@asset
def first(shared_upstream):
return 1
@asset
def second(shared_upstream):
return 2
`;
const result = await parser.parse(content, 'dummy.py');
// We should have two edges, both from the same external source
expect(result.edges.length).toBe(2);
const sources = result.edges.map(e => e.source);
const targets = result.edges.map(e => e.target);
expect(sources).toEqual(['shared_upstream', 'shared_upstream']);
expect(targets).toContain('first');
expect(targets).toContain('second');
// The external source node should only be created once
const externalNodes = result.nodes.filter(n => n.id === 'shared_upstream');
expect(externalNodes).toHaveLength(1);
});
});
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+6
to
+8
| it('should identify Dagster files', () => { | ||
| const content = 'from dagster import asset, Definitions\n@asset\ndef my_asset(): pass'; | ||
| expect(parser.canParse('defs.py', content)).toBe(true); |
There was a problem hiding this comment.
suggestion (testing): Add more canParse coverage for the different Dagster detection heuristics.
This test only covers the from dagster import ... path. Since canParse also detects import dagster, @op, and definitions(, please add tests for each of these cases, plus one where none are present, to prevent regressions in the detection logic.
Suggested implementation:
import { DagsterParser } from '../src/extension/parsers/data-processing/DagsterParser';
describe('DagsterParser', () => {
const parser = new DagsterParser();
describe('canParse', () => {
it('returns true for files using "from dagster import ..."', () => {
const content = 'from dagster import asset, Definitions\n@asset\ndef my_asset(): pass';
expect(parser.canParse('defs.py', content)).toBe(true);
});
it('returns true for files using "import dagster"', () => {
const content = 'import dagster\n\ndef my_job():\n pass';
expect(parser.canParse('job.py', content)).toBe(true);
});
it('returns true for files using "@op" decorator', () => {
const content = 'from dagster import op\n\n@op\ndef my_op(context):\n pass';
expect(parser.canParse('ops.py', content)).toBe(true);
});
it('returns true for files defining "Definitions("', () => {
const content = 'from dagster import Definitions\n\ndefs = Definitions(assets=[], jobs=[])';
expect(parser.canParse('defs.py', content)).toBe(true);
});
it('returns false for non-Dagster Python files', () => {
const content = 'def not_dagster():\n return "hello"';
expect(parser.canParse('script.py', content)).toBe(false);
});
});
it('should parse assets and their dependencies from arguments', async () => {
const content = `If there are other describe('DagsterParser'...) blocks or existing canParse tests elsewhere in this file, you may want to consolidate them into this new describe('canParse', ...) block for clarity, or adjust the test names to avoid duplication.
Comment on lines
+41
to
+56
| it('should parse dependencies from deps argument in decorator', async () => { | ||
| const content = ` | ||
| @asset(deps=["upstream_asset"]) | ||
| def downstream_asset(): | ||
| pass | ||
| `; | ||
| const result = await parser.parse(content, 'dummy.py'); | ||
| expect(result.edges.length).toBe(1); | ||
| expect(result.edges[0]).toEqual({ | ||
| id: 'e-upstream_asset-downstream_asset', | ||
| source: 'upstream_asset', | ||
| target: 'downstream_asset' | ||
| }); | ||
| // Should also detect upstream_asset as an external asset | ||
| expect(result.nodes.find(n => n.id === 'upstream_asset')).toBeDefined(); | ||
| }); |
There was a problem hiding this comment.
suggestion (testing): Extend deps tests to cover multiple deps and AssetKey(...) usage.
The current test only covers a single string dep. To better validate the parsing logic, please also add cases with multiple deps in the list (verifying all edges are created) and with values like AssetKey("foo") to confirm foo is extracted correctly, matching typical Dagster usage.
Suggested change
| it('should parse dependencies from deps argument in decorator', async () => { | |
| const content = ` | |
| @asset(deps=["upstream_asset"]) | |
| def downstream_asset(): | |
| pass | |
| `; | |
| const result = await parser.parse(content, 'dummy.py'); | |
| expect(result.edges.length).toBe(1); | |
| expect(result.edges[0]).toEqual({ | |
| id: 'e-upstream_asset-downstream_asset', | |
| source: 'upstream_asset', | |
| target: 'downstream_asset' | |
| }); | |
| // Should also detect upstream_asset as an external asset | |
| expect(result.nodes.find(n => n.id === 'upstream_asset')).toBeDefined(); | |
| }); | |
| it('should parse dependencies from deps argument in decorator', async () => { | |
| const content = ` | |
| @asset(deps=["upstream_asset"]) | |
| def downstream_asset(): | |
| pass | |
| `; | |
| const result = await parser.parse(content, 'dummy.py'); | |
| expect(result.edges.length).toBe(1); | |
| expect(result.edges[0]).toEqual({ | |
| id: 'e-upstream_asset-downstream_asset', | |
| source: 'upstream_asset', | |
| target: 'downstream_asset', | |
| }); | |
| // Should also detect upstream_asset as an external asset | |
| expect(result.nodes.find((n) => n.id === 'upstream_asset')).toBeDefined(); | |
| }); | |
| it('should parse multiple deps and AssetKey deps in decorator', async () => { | |
| const content = ` | |
| from dagster import asset, AssetKey | |
| @asset(deps=["upstream_asset", "another_upstream", AssetKey("keyed_asset")]) | |
| def downstream_asset(): | |
| pass | |
| `; | |
| const result = await parser.parse(content, 'dummy.py'); | |
| expect(result.edges.length).toBe(3); | |
| expect(result.edges).toEqual( | |
| expect.arrayContaining([ | |
| { | |
| id: 'e-upstream_asset-downstream_asset', | |
| source: 'upstream_asset', | |
| target: 'downstream_asset', | |
| }, | |
| { | |
| id: 'e-another_upstream-downstream_asset', | |
| source: 'another_upstream', | |
| target: 'downstream_asset', | |
| }, | |
| { | |
| id: 'e-keyed_asset-downstream_asset', | |
| source: 'keyed_asset', | |
| target: 'downstream_asset', | |
| }, | |
| ]), | |
| ); | |
| // All upstream assets (including AssetKey-based) should be detected as external assets | |
| ['upstream_asset', 'another_upstream', 'keyed_asset'].forEach((id) => { | |
| expect(result.nodes.find((n) => n.id === id)).toBeDefined(); | |
| }); | |
| }); |
Comment on lines
+58
to
+71
| it('should handle ops and mixed pipelines', async () => { | ||
| const content = ` | ||
| @asset | ||
| def my_asset(): pass | ||
|
|
||
| @op | ||
| def my_op(my_asset): pass | ||
| `; | ||
| const result = await parser.parse(content, 'dummy.py'); | ||
| const op = result.nodes.find(n => n.id === 'my_op'); | ||
| expect(op?.type).toBe('default'); | ||
| expect(result.edges[0].source).toBe('my_asset'); | ||
| expect(result.edges[0].target).toBe('my_op'); | ||
| }); |
There was a problem hiding this comment.
suggestion (testing): Add assertions for node metadata (framework and asset/op classification) in mixed pipelines.
To better cover the visualization behavior, please also assert the node metadata for both nodes (e.g. framework === 'Dagster', data.dataType === 'asset' for the asset and 'op' for the op). This helps catch future parser changes that might break the UI integration.
Suggested change
| it('should handle ops and mixed pipelines', async () => { | |
| const content = ` | |
| @asset | |
| def my_asset(): pass | |
| @op | |
| def my_op(my_asset): pass | |
| `; | |
| const result = await parser.parse(content, 'dummy.py'); | |
| const op = result.nodes.find(n => n.id === 'my_op'); | |
| expect(op?.type).toBe('default'); | |
| expect(result.edges[0].source).toBe('my_asset'); | |
| expect(result.edges[0].target).toBe('my_op'); | |
| }); | |
| it('should handle ops and mixed pipelines', async () => { | |
| const content = ` | |
| @asset | |
| def my_asset(): pass | |
| @op | |
| def my_op(my_asset): pass | |
| `; | |
| const result = await parser.parse(content, 'dummy.py'); | |
| const asset = result.nodes.find(n => n.id === 'my_asset'); | |
| const op = result.nodes.find(n => n.id === 'my_op'); | |
| // Basic node type check | |
| expect(op?.type).toBe('default'); | |
| // Framework and classification metadata | |
| expect(asset?.framework).toBe('Dagster'); | |
| expect(op?.framework).toBe('Dagster'); | |
| expect(asset?.data?.dataType).toBe('asset'); | |
| expect(op?.data?.dataType).toBe('op'); | |
| // Edge wiring between asset and op | |
| expect(result.edges[0].source).toBe('my_asset'); | |
| expect(result.edges[0].target).toBe('my_op'); | |
| }); |
Comment on lines
+73
to
+83
| it('should handle type hints in arguments', async () => { | ||
| const content = ` | ||
| @asset | ||
| def typed_asset(upstream: DataFrame, context: AssetExecutionContext): | ||
| pass | ||
| `; | ||
| const result = await parser.parse(content, 'dummy.py'); | ||
| expect(result.edges.length).toBe(1); | ||
| expect(result.edges[0].source).toBe('upstream'); | ||
| }); | ||
| }); |
There was a problem hiding this comment.
suggestion (testing): Include a test for external-node deduplication when multiple edges reference the same source.
Right now we only cover the single-dependency case. Please also add a test where two downstream assets/ops depend on the same upstream name that isn’t defined locally, and assert that the parser still creates exactly one external node for that shared source. This will verify the Set-based deduplication and guard against unexpected graph inflation.
Suggested change
| it('should handle type hints in arguments', async () => { | |
| const content = ` | |
| @asset | |
| def typed_asset(upstream: DataFrame, context: AssetExecutionContext): | |
| pass | |
| `; | |
| const result = await parser.parse(content, 'dummy.py'); | |
| expect(result.edges.length).toBe(1); | |
| expect(result.edges[0].source).toBe('upstream'); | |
| }); | |
| }); | |
| it('should handle type hints in arguments', async () => { | |
| const content = ` | |
| @asset | |
| def typed_asset(upstream: DataFrame, context: AssetExecutionContext): | |
| pass | |
| `; | |
| const result = await parser.parse(content, 'dummy.py'); | |
| expect(result.edges.length).toBe(1); | |
| expect(result.edges[0].source).toBe('upstream'); | |
| }); | |
| it('should deduplicate external nodes shared by multiple downstream assets', async () => { | |
| const content = ` | |
| from dagster import asset | |
| @asset | |
| def first(shared_upstream): | |
| return 1 | |
| @asset | |
| def second(shared_upstream): | |
| return 2 | |
| `; | |
| const result = await parser.parse(content, 'dummy.py'); | |
| // We should have two edges, both from the same external source | |
| expect(result.edges.length).toBe(2); | |
| const sources = result.edges.map(e => e.source); | |
| const targets = result.edges.map(e => e.target); | |
| expect(sources).toEqual(['shared_upstream', 'shared_upstream']); | |
| expect(targets).toContain('first'); | |
| expect(targets).toContain('second'); | |
| // The external source node should only be created once | |
| const externalNodes = result.nodes.filter(n => n.id === 'shared_upstream'); | |
| expect(externalNodes).toHaveLength(1); | |
| }); | |
| }); |
- Implemented DagsterParser using TypeScript-only regex-based extraction. - Detects Software-Defined Assets, Ops, and their dependencies from Python source. - Supports both function-argument dependencies and explicit 'deps' decorator arguments. - Added visual differentiation between assets (artifact nodes) and ops (default nodes). - Integrated Dagster into the Data Processing pipeline category. - Updated webview UI with fallback icons for Dagster framework detection. - Refined parser for robustness (multiline support, complex signature skipping). - Improved webview component typing and status handling. Co-authored-by: Resmung0 <36093882+Resmung0@users.noreply.github.com>
Co-authored-by: Resmung0 <36093882+Resmung0@users.noreply.github.com>
- Implemented context-aware argument splitting for DagsterParser. - Added multiline decorator and multiple decorator support. - Refactored webview components for improved type safety and consistency. - Narrowed file search patterns for performance. - Expanded test suite to cover complex edge cases. - Resolved PR comments. Co-authored-by: Resmung0 <36093882+Resmung0@users.noreply.github.com>
- Added tests for no-arg assets, varargs (*args/**kwargs), and positional-only markers. - Verified robust argument parsing logic in DagsterParser. - Added tests for multiple dependencies and AssetKey usage in decorators. - Resolved PR comments. Co-authored-by: Resmung0 <36093882+Resmung0@users.noreply.github.com>
- Implemented robust argument splitting for DagsterParser (handles brackets, parentheses). - Added support for multiple decorators and multiline definitions. - Expanded test suite with suggested cases for deps, AssetKey, and argument edge cases. - Improved webview component type safety and artifact rendering consistency. - Resolved all PR comments. Co-authored-by: Resmung0 <36093882+Resmung0@users.noreply.github.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This change adds support for Dagster pipelines to the Caldera extension.
Key improvements:
DagsterParserin TypeScript to extract pipeline graphs (assets and ops) from Python source files using static analysis (regex).PR created automatically by Jules for task 2508395940145616498 started by @Resmung0
Summary by Sourcery
Add Dagster pipeline support to the data processing pipeline system and surface Dagster-specific nodes and metadata in the webview.
New Features:
Enhancements:
Tests: