runs

Documentation

Index

• Constants
• Variables
• type AssetResult
• type CreateAssetInput
• type DestroyRunResponse
• type DocumentationInput
• type DocumentationResult
• type LineageInput
• type LineageResult
• type PostgresRepository
  • func NewPostgresRepository(db *pgxpool.Pool) *PostgresRepository
  • func (r *PostgresRepository) AddCheckpoint(ctx context.Context, runDBID string, checkpoint *plugin.RunCheckpoint) error
  • func (r *PostgresRepository) AddRunEntity(ctx context.Context, runDBID string, entity *RunEntity) error
  • func (r *PostgresRepository) CleanupStaleRuns(ctx context.Context, timeout time.Duration) (int, error)
  • func (r *PostgresRepository) Create(ctx context.Context, run *plugin.Run) error
  • func (r *PostgresRepository) DeleteCheckpoints(ctx context.Context, pipelineName, sourceName string) error
  • func (r *PostgresRepository) Get(ctx context.Context, id string) (*plugin.Run, error)
  • func (r *PostgresRepository) GetByRunID(ctx context.Context, runID string) (*plugin.Run, error)
  • func (r *PostgresRepository) GetLastRunCheckpoints(ctx context.Context, pipelineName, sourceName string) (map[string]*plugin.RunCheckpoint, error)
  • func (r *PostgresRepository) GetPipelines(ctx context.Context) ([]string, error)
  • func (r *PostgresRepository) List(ctx context.Context, pipelineName string, limit, offset int) ([]*plugin.Run, int, error)
  • func (r *PostgresRepository) ListRunEntities(ctx context.Context, runDBID, entityType, status string, limit, offset int) ([]*RunEntity, int, error)
  • func (r *PostgresRepository) ListWithFilters(ctx context.Context, pipelines, statuses []string, limit, offset int) ([]*plugin.Run, int, []string, error)
  • func (r *PostgresRepository) Update(ctx context.Context, run *plugin.Run) error
• type ProcessAssetsResponse
• type Repository
• type RunEntity
• type Service
  • func NewService(repo Repository, assetService asset.Service, lineageService lineage.Service) Service

Constants

const (
	StatusCreated   = "created"
	StatusUpdated   = "updated"
	StatusUnchanged = "unchanged"
	StatusDeleted   = "deleted"
	StatusFailed    = "failed"
)

Variables

var (
	ErrRunNotFound   = errors.New("run not found")
	ErrInvalidInput  = errors.New("invalid input")
	ErrInvalidStatus = errors.New("invalid status transition")
)

var (
	ErrNotFound = errors.New("run not found")
	ErrConflict = errors.New("run already exists")
)

Types

type AssetResult

type AssetResult struct {
	Name     string      `json:"name"`
	Type     string      `json:"type"`
	Provider string      `json:"provider"`
	MRN      string      `json:"mrn"`
	Asset    interface{} `json:"asset"`
	Status   string      `json:"status"`
	Error    string      `json:"error,omitempty"`
}

type CreateAssetInput

type CreateAssetInput struct {
	Name          string                 `json:"name"`
	Type          string                 `json:"type"`
	Providers     []string               `json:"providers"`
	Description   *string                `json:"description"`
	Metadata      map[string]interface{} `json:"metadata"`
	Schema        map[string]interface{} `json:"schema"`
	Tags          []string               `json:"tags"`
	Sources       []string               `json:"sources"`
	ExternalLinks []map[string]string    `json:"external_links"`
}

type DestroyRunResponse

type DestroyRunResponse struct {
	AssetsDeleted        int      `json:"assets_deleted"`
	LineageDeleted       int      `json:"lineage_deleted"`
	DocumentationDeleted int      `json:"documentation_deleted"`
	DeletedEntityMRNs    []string `json:"deleted_entity_mrns"`
}

type DocumentationInput

type DocumentationInput struct {
	AssetMRN string `json:"asset_mrn"`
	Content  string `json:"content"`
	Type     string `json:"type"`
}

type DocumentationResult

type DocumentationResult struct {
	AssetMRN string `json:"asset_mrn"`
	Type     string `json:"type"`
	Status   string `json:"status"`
	Error    string `json:"error,omitempty"`
}

type LineageInput

type LineageInput struct {
	Source string `json:"source"`
	Target string `json:"target"`
	Type   string `json:"type"`
}

type LineageResult

type LineageResult struct {
	Source string `json:"source"`
	Target string `json:"target"`
	Type   string `json:"type"`
	Status string `json:"status"`
	Error  string `json:"error,omitempty"`
}

type PostgresRepository

type PostgresRepository struct {
	// contains filtered or unexported fields
}

func NewPostgresRepository

func NewPostgresRepository(db *pgxpool.Pool) *PostgresRepository

func (*PostgresRepository) AddCheckpoint

func (r *PostgresRepository) AddCheckpoint(ctx context.Context, runDBID string, checkpoint *plugin.RunCheckpoint) error

func (*PostgresRepository) AddRunEntity

func (r *PostgresRepository) AddRunEntity(ctx context.Context, runDBID string, entity *RunEntity) error

func (*PostgresRepository) CleanupStaleRuns

func (r *PostgresRepository) CleanupStaleRuns(ctx context.Context, timeout time.Duration) (int, error)

func (*PostgresRepository) Create

func (r *PostgresRepository) Create(ctx context.Context, run *plugin.Run) error

func (*PostgresRepository) DeleteCheckpoints

func (r *PostgresRepository) DeleteCheckpoints(ctx context.Context, pipelineName, sourceName string) error

func (*PostgresRepository) Get

func (r *PostgresRepository) Get(ctx context.Context, id string) (*plugin.Run, error)

func (*PostgresRepository) GetByRunID

func (r *PostgresRepository) GetByRunID(ctx context.Context, runID string) (*plugin.Run, error)

func (*PostgresRepository) GetLastRunCheckpoints

func (r *PostgresRepository) GetLastRunCheckpoints(ctx context.Context, pipelineName, sourceName string) (map[string]*plugin.RunCheckpoint, error)

func (*PostgresRepository) GetPipelines

func (r *PostgresRepository) GetPipelines(ctx context.Context) ([]string, error)

func (*PostgresRepository) List

func (r *PostgresRepository) List(ctx context.Context, pipelineName string, limit, offset int) ([]*plugin.Run, int, error)

func (*PostgresRepository) ListRunEntities

func (r *PostgresRepository) ListRunEntities(ctx context.Context, runDBID, entityType, status string, limit, offset int) ([]*RunEntity, int, error)

func (*PostgresRepository) ListWithFilters

func (r *PostgresRepository) ListWithFilters(ctx context.Context, pipelines, statuses []string, limit, offset int) ([]*plugin.Run, int, []string, error)

func (*PostgresRepository) Update

func (r *PostgresRepository) Update(ctx context.Context, run *plugin.Run) error

type ProcessAssetsResponse

type ProcessAssetsResponse struct {
	Assets               []AssetResult         `json:"assets"`
	Lineage              []LineageResult       `json:"lineage"`
	Documentation        []DocumentationResult `json:"documentation"`
	StaleEntitiesRemoved []string              `json:"stale_entities_removed,omitempty"`
}

type Repository

type Repository interface {
	Create(ctx context.Context, run *plugin.Run) error
	Get(ctx context.Context, id string) (*plugin.Run, error)
	GetByRunID(ctx context.Context, runID string) (*plugin.Run, error)
	Update(ctx context.Context, run *plugin.Run) error
	List(ctx context.Context, pipelineName string, limit, offset int) ([]*plugin.Run, int, error)
	ListWithFilters(ctx context.Context, pipelines, statuses []string, limit, offset int) ([]*plugin.Run, int, []string, error)
	AddCheckpoint(ctx context.Context, runDBID string, checkpoint *plugin.RunCheckpoint) error
	DeleteCheckpoints(ctx context.Context, pipelineName, sourceName string) error
	GetLastRunCheckpoints(ctx context.Context, pipelineName, sourceName string) (map[string]*plugin.RunCheckpoint, error)
	CleanupStaleRuns(ctx context.Context, timeout time.Duration) (int, error)
	AddRunEntity(ctx context.Context, runDBID string, entity *RunEntity) error
	ListRunEntities(ctx context.Context, runDBID, entityType, status string, limit, offset int) ([]*RunEntity, int, error)
}

type RunEntity

type RunEntity struct {
	ID           string    `json:"id"`
	RunID        string    `json:"run_id"`
	EntityType   string    `json:"entity_type"`
	EntityMRN    string    `json:"entity_mrn"`
	EntityName   string    `json:"entity_name,omitempty"`
	Status       string    `json:"status"`
	ErrorMessage string    `json:"error_message,omitempty"`
	CreatedAt    time.Time `json:"created_at"`
}

type Service

type Service interface {
	StartRun(ctx context.Context, pipelineName, sourceName, createdBy string, config plugin.RawPluginConfig) (*plugin.Run, error)
	CompleteRun(ctx context.Context, runID string, status plugin.RunStatus, summary *plugin.RunSummary, errorMessage string) error
	ProcessAssets(ctx context.Context, runID string, assets []CreateAssetInput, pipelineName, sourceName string) (*ProcessAssetsResponse, error)
	ProcessEntities(ctx context.Context, runID string, assets []CreateAssetInput, lineage []LineageInput, docs []DocumentationInput, pipelineName, sourceName string) (*ProcessAssetsResponse, error)
	AddCheckpoint(ctx context.Context, runID, entityType, entityMRN, operation string, sourceFields []string) error
	GetLastRunCheckpoints(ctx context.Context, pipelineName, sourceName string) (map[string]*plugin.RunCheckpoint, error)
	GetStaleEntities(ctx context.Context, lastCheckpoints map[string]*plugin.RunCheckpoint, currentEntityMRNs []string) []string
	DestroyPipeline(ctx context.Context, pipelineName string) (*DestroyRunResponse, error)
	CleanupStaleRuns(ctx context.Context, timeout time.Duration) (int, error)
	ListRuns(ctx context.Context, pipelineName string, limit, offset int) ([]*plugin.Run, int, error)
	ListRunsWithFilters(ctx context.Context, pipelines, statuses []string, limit, offset int) ([]*plugin.Run, int, []string, error)
	GetRun(ctx context.Context, id string) (*plugin.Run, error)
	ListRunEntities(ctx context.Context, runID, entityType, status string, limit, offset int) ([]*RunEntity, int, error)
}

func NewService

func NewService(repo Repository, assetService asset.Service, lineageService lineage.Service) Service