Covers DAG design principles (idempotent, atomic, incremental, observable) with task dependency patterns for linear, fan-out, fan-in, and complex workflows
Includes TaskFlow API decorators for cleaner code with automatic XCom passing, dynamic DAG generation from configs, and branching with conditional logic
Provides sensor patterns for S3 files, external task dependencies, and custom senso
Fetches airflow-dag-patterns from wshobson/agents and configures it for Cursor.
3
Select Cursor when prompted
The CLI shows a list of agents. Use arrow keys and space to select Cursor:
◆ Which agents do you want to install to?
│
│ ── Universal (.agents/skills) ────────────────
│ · Cline · Codex · Goose · Windsurf
│ ●Cursor(selected)
│ · Cursor · Aider · Continue
4
Verify installation
Confirm successful installation by checking the skill directory location:
.cursor/skills/airflow-dag-patterns
Restart Cursor to activate airflow-dag-patterns. Access via /airflow-dag-patterns in your agent's command palette.
⚠
Security Notice
We perform automated surface-level scans (Gen AI Scanner, Socket, Snyk) during installation. These checks detect common vulnerabilities but do not guarantee complete security. Always review skill source code and verify the publisher's reputation before production use.
Skills execute code in your environment. Always review source, verify the publisher, and test in isolation before production.
# dags/example_dag.pyfrom datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
default_args ={'owner':'data-team','depends_on_past':False,'email_on_failure':True,'email_on_retry':False,'retries':3,'retry_delay': timedelta(minutes=5),'retry_exponential_backoff':True,'max_retry_delay': timedelta(hours=1),}with DAG( dag_id='example_etl', default_args=default_args, description='Example ETL pipeline', schedule='0 6 * * *',# Daily at 6 AM start_date=datetime(2024,1,1), catchup=False, tags=['etl','example'], max_active_runs=1,)as dag: start = EmptyOperator(task_id='start')defextract_data(**context): execution_date = context['ds']# Extract logic herereturn{'records':1000} extract = PythonOperator( task_id='extract', python_callable=extract_data,) end = EmptyOperator(task_id='end') start >> extract >> end
Patterns
Pattern 1: TaskFlow API (Airflow 2.0+)
# dags/taskflow_example.pyfrom datetime import datetime
from airflow.decorators import dag, task
from airflow.models import Variable
@dag( dag_id='taskflow_etl', schedule='@daily', start_date=datetime(2024,1,1), catchup=False, tags=['etl','taskflow'],)deftaskflow_etl():"""ETL pipeline using TaskFlow API"""@task()defextract(source:str)->dict:"""Extract data from source"""import pandas as pd
df = pd.read_csv(f's3://bucket/{source}/{{ ds }}.csv')return{'data': df.to_dict(),'rows':len(df)}@task()deftransform(extracted:dict)->dict:"""Transform extracted data"""import pandas as pd
df = pd.DataFrame(extracted['data']) df['processed_at']= datetime.now() df = df.dropna()return{'data': df.to_dict(),'rows':len(df)}@task()defload(transformed:dict, target:str):"""Load data to target"""import pandas as pd
df = pd.DataFrame(transformed['data']) df.to_parquet(f's3://bucket/{target}/{{ ds }}.parquet')return transformed['rows']@task()defnotify(rows_loaded:int):"""Send notification"""print(f'Loaded {rows_loaded} rows')# Define dependencies with XCom passing extracted = extract(source='raw_data') transformed = transform(extracted) loaded = load(transformed, target='processed_data') notify(loaded)# Instantiate the DAGtaskflow_etl()
Pattern 2: Dynamic DAG Generation
# dags/dynamic_dag_factory.pyfrom datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
import json
# Configuration for multiple similar pipelinesPIPELINE_CONFIGS =[{'name':'customers','schedule':'@daily','source':'s3://raw/customers'},{'name':'orders','schedule':'@hourly','source':'s3://raw/orders'},{'name':'products','schedule':
Implementation Guide
Prerequisites
›Claude Desktop or compatible AI client with skill support
›Clear understanding of task or problem to solve
›Willingness to iterate and refine outputs
Time Estimate
15-45 minutes depending on use case complexity
Steps
1Install skill using provided installation command
2Test with simple use case relevant to your work
3Evaluate output quality and relevance
4Iterate on prompts to improve results
5Integrate into regular workflow if valuable
Common Pitfalls
⚠Expecting perfect results without iteration
⚠Not providing enough context in prompts
⚠Using skill for tasks outside its intended scope
⚠Accepting outputs without review and validation
Best Practices
✓ Do
+Start with clear, specific prompts
+Provide relevant context and constraints
+Review and refine all outputs before using
+Iterate to improve output quality
+Document successful prompt patterns
✗ Don't
−Don't use without understanding skill limitations
−Don't skip validation of outputs
−Don't share sensitive information in prompts
−Don't expect skill to replace human judgment
💡 Pro Tips
★Be specific about desired format and style
★Ask for multiple options to choose from
★Request explanations to understand reasoning
★Combine AI efficiency with human expertise
When to Use This
✓ Use when
Use when skill capabilities match your task, clear ROI on time saved, and you can validate outputs. Best for repetitive tasks, learning, and quality improvement.
✗ Avoid when
Avoid when task requires deep expertise you can't validate, involves sensitive decisions, or when learning process is more valuable than speed of completion.
Learning Path
1Familiarize yourself with skill capabilities and limitations
2Start with low-risk, non-critical tasks
3Progress to more complex and valuable use cases
4Build expertise through regular use and experimentation
