cnolanminich

import json
import textwrap
from typing import Any, Mapping, List, Tuple
from dagster import (
    AutomationCondition,
    AssetKey,
    BackfillPolicy,
    DailyPartitionsDefinition,
    job,
    op,

cnolanminich

with events as (select distinct

    DATE_FORMAT(timestamp, '%Y-%m')  as event_month,
    dagster_event_type,
    coalesce(run_id, '||', step_key) as step_id,
    count(1) as credits

from event_logs
where dagster_event_type = 'STEP_START'

cnolanminich

with seconds_in_day as (
  select generate_series(0, 86400 - 1) as sec
)
, base_days as (
select generate_series(current_date - 7,current_date,'1 day'::interval) as days
),
days_and_seconds as (
  select
  		base_days.days,
    	base_days.days + (seconds_in_day.sec || ' seconds')::interval as start_bucket,

cnolanminich

# runs at 5 minutes past the hour for demonstration purposes.
daily_success_condition = (
    dg.AutomationCondition.newly_updated()
    .since(dg.AutomationCondition.on_cron("*/5 * * * *"))
)

custom_condition = (
    dg.AutomationCondition.on_cron("*/5 * * * *")  # Runs at 9 AM daily
    | (
        dg.AutomationCondition.any_deps_updated()  # When any dependency updates

cnolanminich

# hooli-ml/databricks_mlops/dagster_pipeline.py
import dagster as dg
from hooli_ml.defs.resources import get_env, MLWorkflowConfig, DatabricksResource
from databricks.sdk.service import jobs

NOTEBOOK_ROOT_PATH = "/Users/christian@dagsterlabs.com/.bundle/databricks_mlops/dev/files/"

@dg.asset(kinds={"databricks", "feature_engineering"})
def feature_engineering_pickup(context: dg.AssetExecutionContext, databricks_resource: DatabricksResource):
    run_id = databricks_resource.run_and_stream_notebook_logs_sdk(

cnolanminich

with daily as(
select 
    ds::date as reporting_date,
    sum(standard_credits) as daily_standard_credits,
    sum(high_velocity_credits) as daily_high_velo_credits,
    daily_standard_credits + daily_high_velo_credits as total_credits_daily
from purina.product.usage_metrics_daily 
where usage_metrics_daily.organization_id = {org_id here} 
group by ds::date
)

cnolanminich

pipeline {
    agent any
    
    environment {
        // Dagster Cloud Configuration
        DAGSTER_CLOUD_ORGANIZATION = 'your-org-name'  // Replace with your Dagster+ org name
        DAGSTER_CLOUD_API_TOKEN = credentials('dagster-cloud-api-token')  // Jenkins credential ID
        DAGSTER_BUILD_STATEDIR = "${WORKSPACE}/build_state"
        
        // AWS ECR Configuration

cnolanminich

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.settings import Session
from airflow.models import DagRun
from datetime import timedelta
import pandas as pd
import numpy as np

# Constants for flexibility

cnolanminich

Dagster Demo Generator - GitHub Copilot Guide

This document provides comprehensive instructions for GitHub Copilot to replicate the Dagster demo generation workflow implemented in this repository.

Architecture Overview

The Dagster Demo Generator automates the creation of demonstration Dagster projects using a multi-tiered workflow:

1. Command Entry Point: /create-dagster-demo slash command (.claude/commands/create-dagster-demo-2.md)
2. Skills System: Three specialized skills that handle different aspects of Dagster project creation

cnolanminich

mutation {
  createSecret(
    secretName: "MY_API_KEY"
    secretValue: "abc123"
    scopes: { fullDeploymentScope: true }
    locationNames: ["prod"]
  ) {
    __typename
    ... on Error {
      message