b7e07db9-8532-4cc5-bce5-55e428d24888

Committed 23 Mar 2026 04:12PM UTC coverage: 36.048% (+0.2%) from 35.862%

Build # b7e07db9-8532-4cc5-bce5-55e428d24888

Build Type

Pull #2042

circleci

Committed by

kaxil

Commit Message

Add `astro api registry` command for querying the Airflow Provider Registry

Adds a new `astro api registry` command that provides access to the public
Airflow Provider Registry (https://airflow.apache.org/registry), following
the same patterns as `astro api airflow` and `astro api cloud`.

Supports both path-based and operation ID-based queries:
  astro api registry /providers.json
  astro api registry getProviderModulesLatest -p providerId=amazon
  astro api registry getProviderModulesByVersion -p providerId=amazon -p version=9.22.0
  astro api registry ls
  astro api registry describe getProviderModulesLatest

Uses the same base flags and infrastructure as airflow/cloud: -X, -F, -f,
-H, -p, -i, -q, -t, --silent, --verbose, --generate.

Reuses the shared RequestOptions, executeRequest, resolveOperationID,
applyPathParams, and openapi.Cache infrastructure. No new dependencies.

Pull Request Pull Request #2042: Add `astro api registry` command

Coverage Stats

217 of 258 new or added lines in 3 files covered. (84.11%)

24452 of 67832 relevant lines covered (36.05%)

8.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.45

/pkg/openapi/cache.go

package openapi

import (
        "context"
        "encoding/json"
        "fmt"
        "io"
        "net/http"
        "os"
        "path/filepath"
        "strings"
        "time"

        "github.com/astronomer/astro-cli/config"
        v3high "github.com/pb33f/libopenapi/datamodel/high/v3"
)

const (
        // SpecURL is the URL to fetch the Astro Cloud API OpenAPI specification.
        SpecURL = "https://api.astronomer.io/spec/v1.0"
        // CloudCacheFileName is the name of the cache file for the Astro Cloud API.
        CloudCacheFileName = "openapi-cache.json"
        // AirflowCacheFileNameTemplate is the template for version-specific Airflow cache files.
        AirflowCacheFileNameTemplate = "openapi-airflow-%s-cache.json"
        // CacheTTL is how long the cache is valid.
        CacheTTL = 24 * time.Hour
        // FetchTimeout is the timeout for fetching the spec.
        FetchTimeout = 30 * time.Second
        // dirPermissions is the permission mode for created directories.
        dirPermissions = 0o755
        // filePermissions is the permission mode for the cache file.
        filePermissions = 0o600
)

// AirflowCacheFileNameForVersion returns the cache file name for a specific Airflow version.
// The version is normalized to strip build metadata (e.g., "3.1.7+astro.1" -> "3.1.7").
func AirflowCacheFileNameForVersion(version string) string {
        normalizedVersion := NormalizeAirflowVersion(version)
        return fmt.Sprintf(AirflowCacheFileNameTemplate, normalizedVersion)
}

// CloudCacheFileNameForDomain returns the cache file name for a specific domain.
// For the default domain (astronomer.io), it returns the standard cache file name
// for backward compatibility. For other domains, a domain-qualified name is used
// to avoid cache collisions between environments.
func CloudCacheFileNameForDomain(domain string) string {
        if domain == "" || domain == "astronomer.io" {
                return CloudCacheFileName
        }
        normalized := strings.ReplaceAll(domain, ".", "_")
        return fmt.Sprintf("openapi-cache-%s.json", normalized)
}

// CachedSpec wraps raw spec data with metadata for caching.
// RawSpec is stored as []byte which json.Marshal encodes as base64.
type CachedSpec struct {
        RawSpec   []byte    `json:"rawSpec"`
        FetchedAt time.Time `json:"fetchedAt"`
}

// Cache manages fetching and caching of an OpenAPI specification.
type Cache struct {
        specURL     string
        cachePath   string
        stripPrefix string // optional prefix to strip from endpoint paths (e.g. "/api/v2")
        httpClient  *http.Client
        doc         *v3high.Document
        rawSpec     []byte // raw spec bytes for cache serialization
        fetchedAt   time.Time
}

// getHTTPClient returns the configured HTTP client, defaulting to http.DefaultClient.
func (c *Cache) getHTTPClient() *http.Client {
        if c.httpClient != nil {
                return c.httpClient
        }
        return http.DefaultClient
}

// SetHTTPClient configures the HTTP client used to fetch specs.
func (c *Cache) SetHTTPClient(client *http.Client) {
        c.httpClient = client
}

// SetStripPrefix configures a path prefix to strip from endpoint paths
// (e.g. "/api") so callers work with shorter paths.
func (c *Cache) SetStripPrefix(prefix string) {
        c.stripPrefix = prefix
}

// NewCache creates a new OpenAPI cache with default settings.
func NewCache() *Cache {
        return &Cache{
                specURL:   SpecURL,
                cachePath: filepath.Join(config.HomeConfigPath, CloudCacheFileName),
        }
}

// NewCacheWithOptions creates a new OpenAPI cache with custom settings.
func NewCacheWithOptions(specURL, cachePath string) *Cache {
        return &Cache{
                specURL:   specURL,
                cachePath: cachePath,
        }
}

// NewAirflowCacheForVersion creates a new OpenAPI cache configured for a specific Airflow version.
// It automatically determines the correct spec URL and cache file path based on the version.
// Endpoint paths are stripped of their API version prefix (/api/v1 or /api/v2) so that
// callers work with version-agnostic paths like /dags instead of /api/v2/dags.
func NewAirflowCacheForVersion(version string) (*Cache, error) {
        specURL, err := BuildAirflowSpecURL(version)
        if err != nil {
                return nil, fmt.Errorf("building spec URL for version %s: %w", version, err)
        }

        cachePath := filepath.Join(config.HomeConfigPath, AirflowCacheFileNameForVersion(version))

        // Determine the API prefix to strip based on major version
        prefix := "/api/v2"
        normalized := NormalizeAirflowVersion(version)
        if strings.HasPrefix(normalized, "2.") {
                prefix = "/api/v1"
        }

        return &Cache{
                specURL:     specURL,
                cachePath:   cachePath,
                stripPrefix: prefix,
        }, nil
}

// Load loads the OpenAPI spec, using cache if valid or fetching if needed.
// If forceRefresh is true, the cache is ignored and a fresh spec is fetched.
func (c *Cache) Load(forceRefresh bool) error {
        if !forceRefresh {
                // Try to load from cache first
                if err := c.readCache(); err == nil && !c.isExpired() {
                        return nil
                }
        }

        // Fetch fresh spec
        if err := c.fetchSpec(); err != nil {
                // If fetch fails and we have a stale cache, use it
                if c.doc != nil {
                        return nil
                }
                return err
        }

        // Save to cache
        return c.saveCache()
}

// GetSpecURL returns the URL used to fetch the OpenAPI spec.
func (c *Cache) GetSpecURL() string {
        return c.specURL
}

// GetDoc returns the loaded OpenAPI document.
func (c *Cache) GetDoc() *v3high.Document {
        return c.doc
}

// GetEndpoints extracts all endpoints from the loaded spec.
// If a stripPrefix is configured, it is removed from each endpoint path.
func (c *Cache) GetEndpoints() []Endpoint {
        if c.doc == nil {
                return nil
        }
        endpoints := ExtractEndpoints(c.doc)
        if c.stripPrefix != "" {
                for i := range endpoints {
                        endpoints[i].Path = strings.TrimPrefix(endpoints[i].Path, c.stripPrefix)
                }
        }
        return endpoints
}

// IsLoaded returns true if a spec has been loaded.
func (c *Cache) IsLoaded() bool {
        return c.doc != nil
}

// readCache attempts to read the cached spec from disk.
func (c *Cache) readCache() error {
        data, err := os.ReadFile(c.cachePath)
        if err != nil {
                return err
        }

        var cached CachedSpec
        if err := json.Unmarshal(data, &cached); err != nil {
                return err
        }

        doc, err := parseSpec(cached.RawSpec)
        if err != nil {
                return err
        }

        c.doc = doc
        c.rawSpec = cached.RawSpec
        c.fetchedAt = cached.FetchedAt
        return nil
}

// saveCache saves the current spec to disk.
func (c *Cache) saveCache() error {
        cached := CachedSpec{
                RawSpec:   c.rawSpec,
                FetchedAt: c.fetchedAt,
        }

        data, err := json.Marshal(cached)
        if err != nil {
                return err
        }

        // Ensure directory exists
        dir := filepath.Dir(c.cachePath)
        if err := os.MkdirAll(dir, dirPermissions); err != nil {
                return err
        }

        return os.WriteFile(c.cachePath, data, filePermissions)
}

// isExpired returns true if the cached spec has expired.
func (c *Cache) isExpired() bool {
        return time.Since(c.fetchedAt) > CacheTTL
}

// fetchSpec fetches the OpenAPI spec from the remote URL.
func (c *Cache) fetchSpec() error {
        ctx, cancel := context.WithTimeout(context.Background(), FetchTimeout)
        defer cancel()

        req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.specURL, http.NoBody)
        if err != nil {
                return fmt.Errorf("creating request: %w", err)
        }

        // Accept both JSON and YAML
        req.Header.Set("Accept", "application/json, application/yaml, text/yaml, */*")

        resp, err := c.getHTTPClient().Do(req)
        if err != nil {
                return fmt.Errorf("fetching OpenAPI spec: %w", err)
        }
        defer resp.Body.Close()

        if resp.StatusCode != http.StatusOK {
                return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
        }

        body, err := io.ReadAll(resp.Body)
        if err != nil {
                return fmt.Errorf("reading response body: %w", err)
        }

        doc, err := parseSpec(body)
        if err != nil {
                return fmt.Errorf("parsing OpenAPI spec: %w", err)
        }

        c.doc = doc
        c.rawSpec = body
        c.fetchedAt = time.Now()
        return nil
}

// parseSpec parses raw bytes into an OpenAPI v3 document using libopenapi.
// This handles both JSON and YAML formats and resolves $ref references.
func parseSpec(data []byte) (*v3high.Document, error) {
        doc, err := newDocument(data)
        if err != nil {
                return nil, fmt.Errorf("creating document: %w", err)
        }
        model, err := doc.BuildV3Model()
        if err != nil {
                return nil, fmt.Errorf("building v3 model: %w", err)
        }
        if model == nil {
                return nil, fmt.Errorf("building v3 model: unknown error")
        }
        return &model.Model, nil
}

// ClearCache removes the cached spec file.
func (c *Cache) ClearCache() error {
        return os.Remove(c.cachePath)
}

1	package openapi
2
3	import (
4	"context"
5	"encoding/json"
6	"fmt"
7	"io"
8	"net/http"
9	"os"
10	"path/filepath"
11	"strings"
12	"time"
13
14	"github.com/astronomer/astro-cli/config"
15	v3high "github.com/pb33f/libopenapi/datamodel/high/v3"
16	)
17
18	const (
19	// SpecURL is the URL to fetch the Astro Cloud API OpenAPI specification.
20	SpecURL = "https://api.astronomer.io/spec/v1.0"
21	// CloudCacheFileName is the name of the cache file for the Astro Cloud API.
22	CloudCacheFileName = "openapi-cache.json"
23	// AirflowCacheFileNameTemplate is the template for version-specific Airflow cache files.
24	AirflowCacheFileNameTemplate = "openapi-airflow-%s-cache.json"
25	// CacheTTL is how long the cache is valid.
26	CacheTTL = 24 * time.Hour
27	// FetchTimeout is the timeout for fetching the spec.
28	FetchTimeout = 30 * time.Second
29	// dirPermissions is the permission mode for created directories.
30	dirPermissions = 0o755
31	// filePermissions is the permission mode for the cache file.
32	filePermissions = 0o600
33	)
34
35	// AirflowCacheFileNameForVersion returns the cache file name for a specific Airflow version.
36	// The version is normalized to strip build metadata (e.g., "3.1.7+astro.1" -> "3.1.7").
37	func AirflowCacheFileNameForVersion(version string) string {	7✔
38	normalizedVersion := NormalizeAirflowVersion(version)	7✔
39	return fmt.Sprintf(AirflowCacheFileNameTemplate, normalizedVersion)	7✔
40	}	7✔
41
42	// CloudCacheFileNameForDomain returns the cache file name for a specific domain.
43	// For the default domain (astronomer.io), it returns the standard cache file name
44	// for backward compatibility. For other domains, a domain-qualified name is used
45	// to avoid cache collisions between environments.
46	func CloudCacheFileNameForDomain(domain string) string {	7✔
47	if domain == "" \|\| domain == "astronomer.io" {	9✔
48	return CloudCacheFileName	2✔
49	}	2✔
50	normalized := strings.ReplaceAll(domain, ".", "_")	5✔
51	return fmt.Sprintf("openapi-cache-%s.json", normalized)	5✔
52	}
53
54	// CachedSpec wraps raw spec data with metadata for caching.
55	// RawSpec is stored as []byte which json.Marshal encodes as base64.
56	type CachedSpec struct {
57	RawSpec []byte `json:"rawSpec"`
58	FetchedAt time.Time `json:"fetchedAt"`
59	}
60
61	// Cache manages fetching and caching of an OpenAPI specification.
62	type Cache struct {
63	specURL string
64	cachePath string
65	stripPrefix string // optional prefix to strip from endpoint paths (e.g. "/api/v2")
66	httpClient *http.Client
67	doc *v3high.Document
68	rawSpec []byte // raw spec bytes for cache serialization
69	fetchedAt time.Time
70	}
71
72	// getHTTPClient returns the configured HTTP client, defaulting to http.DefaultClient.
73	func (c Cache) getHTTPClient() http.Client {	15✔
74	if c.httpClient != nil {	16✔
75	return c.httpClient	1✔
76	}	1✔
77	return http.DefaultClient	14✔
78	}
79
80	// SetHTTPClient configures the HTTP client used to fetch specs.
81	func (c Cache) SetHTTPClient(client http.Client) {	1✔
82	c.httpClient = client	1✔
83	}	1✔
84
85	// SetStripPrefix configures a path prefix to strip from endpoint paths
86	// (e.g. "/api") so callers work with shorter paths.
NEW 87	func (c *Cache) SetStripPrefix(prefix string) {	×
NEW 88	c.stripPrefix = prefix	×
NEW 89	}	×
90
91	// NewCache creates a new OpenAPI cache with default settings.
92	func NewCache() *Cache {	1✔
93	return &Cache{	1✔
94	specURL: SpecURL,	1✔
95	cachePath: filepath.Join(config.HomeConfigPath, CloudCacheFileName),	1✔
96	}	1✔
97	}	1✔
98
99	// NewCacheWithOptions creates a new OpenAPI cache with custom settings.
100	func NewCacheWithOptions(specURL, cachePath string) *Cache {	5✔
101	return &Cache{	5✔
102	specURL: specURL,	5✔
103	cachePath: cachePath,	5✔
104	}	5✔
105	}	5✔
106
107	// NewAirflowCacheForVersion creates a new OpenAPI cache configured for a specific Airflow version.
108	// It automatically determines the correct spec URL and cache file path based on the version.
109	// Endpoint paths are stripped of their API version prefix (/api/v1 or /api/v2) so that
110	// callers work with version-agnostic paths like /dags instead of /api/v2/dags.
111	func NewAirflowCacheForVersion(version string) (*Cache, error) {	4✔
112	specURL, err := BuildAirflowSpecURL(version)	4✔
113	if err != nil {	5✔
114	return nil, fmt.Errorf("building spec URL for version %s: %w", version, err)	1✔
115	}	1✔
116
117	cachePath := filepath.Join(config.HomeConfigPath, AirflowCacheFileNameForVersion(version))	3✔
118		3✔
119	// Determine the API prefix to strip based on major version	3✔
120	prefix := "/api/v2"	3✔
121	normalized := NormalizeAirflowVersion(version)	3✔
122	if strings.HasPrefix(normalized, "2.") {	4✔
123	prefix = "/api/v1"	1✔
124	}	1✔
125
126	return &Cache{	3✔
127	specURL: specURL,	3✔
128	cachePath: cachePath,	3✔
129	stripPrefix: prefix,	3✔
130	}, nil	3✔
131	}
132
133	// Load loads the OpenAPI spec, using cache if valid or fetching if needed.
134	// If forceRefresh is true, the cache is ignored and a fresh spec is fetched.
135	func (c *Cache) Load(forceRefresh bool) error {	7✔
136	if !forceRefresh {	13✔
137	// Try to load from cache first	6✔
138	if err := c.readCache(); err == nil && !c.isExpired() {	8✔
139	return nil	2✔
140	}	2✔
141	}
142
143	// Fetch fresh spec
144	if err := c.fetchSpec(); err != nil {	6✔
145	// If fetch fails and we have a stale cache, use it	1✔
146	if c.doc != nil {	1✔
147	return nil	×
148	}	×
149	return err	1✔
150	}
151
152	// Save to cache
153	return c.saveCache()	4✔
154	}
155
156	// GetSpecURL returns the URL used to fetch the OpenAPI spec.
157	func (c *Cache) GetSpecURL() string {	×
158	return c.specURL	×
159	}	×
160
161	// GetDoc returns the loaded OpenAPI document.
162	func (c Cache) GetDoc() v3high.Document {	4✔
163	return c.doc	4✔
164	}	4✔
165
166	// GetEndpoints extracts all endpoints from the loaded spec.
167	// If a stripPrefix is configured, it is removed from each endpoint path.
168	func (c *Cache) GetEndpoints() []Endpoint {	3✔
169	if c.doc == nil {	4✔
170	return nil	1✔
171	}	1✔
172	endpoints := ExtractEndpoints(c.doc)	2✔
173	if c.stripPrefix != "" {	3✔
174	for i := range endpoints {	3✔
175	endpoints[i].Path = strings.TrimPrefix(endpoints[i].Path, c.stripPrefix)	2✔
176	}	2✔
177	}
178	return endpoints	2✔
179	}
180
181	// IsLoaded returns true if a spec has been loaded.
182	func (c *Cache) IsLoaded() bool {	2✔
183	return c.doc != nil	2✔
184	}	2✔
185
186	// readCache attempts to read the cached spec from disk.
187	func (c *Cache) readCache() error {	9✔
188	data, err := os.ReadFile(c.cachePath)	9✔
189	if err != nil {	14✔
190	return err	5✔
191	}	5✔
192
193	var cached CachedSpec	4✔
194	if err := json.Unmarshal(data, &cached); err != nil {	5✔
195	return err	1✔
196	}	1✔
197
198	doc, err := parseSpec(cached.RawSpec)	3✔
199	if err != nil {	3✔
200	return err	×
201	}	×
202
203	c.doc = doc	3✔
204	c.rawSpec = cached.RawSpec	3✔
205	c.fetchedAt = cached.FetchedAt	3✔
206	return nil	3✔
207	}
208
209	// saveCache saves the current spec to disk.
210	func (c *Cache) saveCache() error {	5✔
211	cached := CachedSpec{	5✔
212	RawSpec: c.rawSpec,	5✔
213	FetchedAt: c.fetchedAt,	5✔
214	}	5✔
215		5✔
216	data, err := json.Marshal(cached)	5✔
217	if err != nil {	5✔
218	return err	×
219	}	×
220
221	// Ensure directory exists
222	dir := filepath.Dir(c.cachePath)	5✔
223	if err := os.MkdirAll(dir, dirPermissions); err != nil {	5✔
224	return err	×
225	}	×
226
227	return os.WriteFile(c.cachePath, data, filePermissions)	5✔
228	}
229
230	// isExpired returns true if the cached spec has expired.
231	func (c *Cache) isExpired() bool {	5✔
232	return time.Since(c.fetchedAt) > CacheTTL	5✔
233	}	5✔
234
235	// fetchSpec fetches the OpenAPI spec from the remote URL.
236	func (c *Cache) fetchSpec() error {	13✔
237	ctx, cancel := context.WithTimeout(context.Background(), FetchTimeout)	13✔
238	defer cancel()	13✔
239		13✔
240	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.specURL, http.NoBody)	13✔
241	if err != nil {	13✔
242	return fmt.Errorf("creating request: %w", err)	×
243	}	×
244
245	// Accept both JSON and YAML
246	req.Header.Set("Accept", "application/json, application/yaml, text/yaml, /")	13✔
247		13✔
248	resp, err := c.getHTTPClient().Do(req)	13✔
249	if err != nil {	13✔
250	return fmt.Errorf("fetching OpenAPI spec: %w", err)	×
251	}	×
252	defer resp.Body.Close()	13✔
253		13✔
254	if resp.StatusCode != http.StatusOK {	15✔
255	return fmt.Errorf("unexpected status code: %d", resp.StatusCode)	2✔
256	}	2✔
257
258	body, err := io.ReadAll(resp.Body)	11✔
259	if err != nil {	11✔
260	return fmt.Errorf("reading response body: %w", err)	×
261	}	×
262
263	doc, err := parseSpec(body)	11✔
264	if err != nil {	13✔
265	return fmt.Errorf("parsing OpenAPI spec: %w", err)	2✔
266	}	2✔
267
268	c.doc = doc	9✔
269	c.rawSpec = body	9✔
270	c.fetchedAt = time.Now()	9✔
271	return nil	9✔
272	}
273
274	// parseSpec parses raw bytes into an OpenAPI v3 document using libopenapi.
275	// This handles both JSON and YAML formats and resolves $ref references.
276	func parseSpec(data []byte) (*v3high.Document, error) {	23✔
277	doc, err := newDocument(data)	23✔
278	if err != nil {	25✔
279	return nil, fmt.Errorf("creating document: %w", err)	2✔
280	}	2✔
281	model, err := doc.BuildV3Model()	21✔
282	if err != nil {	21✔
283	return nil, fmt.Errorf("building v3 model: %w", err)	×
284	}	×
285	if model == nil {	21✔
286	return nil, fmt.Errorf("building v3 model: unknown error")	×
287	}	×
288	return &model.Model, nil	21✔
289	}
290
291	// ClearCache removes the cached spec file.
292	func (c *Cache) ClearCache() error {	2✔
293	return os.Remove(c.cachePath)	2✔
294	}	2✔

astronomer / astro-cli / b7e07db9-8532-4cc5-bce5-55e428d24888

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous