jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

getandlist

Browse Source
This commit is contained in:
Michael McGuinness
2025-01-03 14:08:04 +00:00
parent 788b21594c
commit 483a47e4e0
8 changed files with 129 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/queries/query.sql
+12 -1
View File
@@ -5,7 +5,7 @@ SELECT id, config FROM queryConfigs where queryId = $1 and addedVersion >= $2 an
SELECT type FROM queries where id = $1;
-- name: DeprecateQuery :exec
INSERT INTO queryDeprecations (queryId) VALUES ($1) RETURNING id;
INSERT INTO queryDeprecations (queryId) VALUES ($1);
-- name: IsQueryDeprecated :one
SELECT EXISTS (
@@ -22,4 +22,15 @@ SELECT q.id, q.type, q.activeVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS
and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion
and r.addedVersion >= q.activeVersion
and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion
GROUP BY q.id, c.config;
-- name: ListQueries :many
SELECT q.id, q.type, q.activeVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds
FROM queries AS q
JOIN queryConfigs AS c ON q.id = c.queryId
JOIN requiredQueries AS r ON q.id = r.queryId
WHERE c.addedVersion >= q.activeVersion
and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion
and r.addedVersion >= q.activeVersion
and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion
GROUP BY q.id, c.config;
internal/database/repository/query.sql.go
+47 -1
View File
@@ -12,7 +12,7 @@ import (
)
const deprecateQuery = `-- name: DeprecateQuery :exec
INSERT INTO queryDeprecations (queryId) VALUES ($1) RETURNING id
INSERT INTO queryDeprecations (queryId) VALUES ($1)
`
func (q *Queries) DeprecateQuery(ctx context.Context, queryid pgtype.UUID) error {
@@ -98,3 +98,49 @@ func (q *Queries) IsQueryDeprecated(ctx context.Context, queryid pgtype.UUID) (b
err := row.Scan(&exists)
return exists, err
}
const listQueries = `-- name: ListQueries :many
SELECT q.id, q.type, q.activeVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds
FROM queries AS q
JOIN queryConfigs AS c ON q.id = c.queryId
JOIN requiredQueries AS r ON q.id = r.queryId
WHERE c.addedVersion >= q.activeVersion
and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion
and r.addedVersion >= q.activeVersion
and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion
GROUP BY q.id, c.config
`
type ListQueriesRow struct {
ID pgtype.UUID
Type Querytype
Activeversion int32
Config []byte
Requiredids []pgtype.UUID
}
func (q *Queries) ListQueries(ctx context.Context) ([]ListQueriesRow, error) {
rows, err := q.db.Query(ctx, listQueries)
if err != nil {
return nil, err
}
defer rows.Close()
var items []ListQueriesRow
for rows.Next() {
var i ListQueriesRow
if err := rows.Scan(
&i.ID,
&i.Type,
&i.Activeversion,
&i.Config,
&i.Requiredids,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
internal/query/list.go
+18 -3
View File
@@ -2,9 +2,24 @@ package query
import (
"context"
queryprocessor "queryorchestration/internal/queryProcessor"
)
func (s *Service) List(ctx context.Context, filters ListFilters) (*[]Query, error) {
// TODO
return nil, nil
func (s *Service) List(ctx context.Context, filters ListFilters) (*[]queryprocessor.Query, error) {
dbQueries, err := s.db.ListQueries(ctx)
if err != nil {
return nil, err
}
queries := make([]queryprocessor.Query, len(dbQueries))
for index, query := range dbQueries {
q, err := queryprocessor.ParseDBListQuery(&query)
if err != nil {
return nil, err
}
queries[index] = *q
}
return &queries, nil
}
internal/query/service.go
+5 -27
View File
@@ -9,20 +9,14 @@ import (
"github.com/google/uuid"
)
type Query struct {
ID uuid.UUID
Type queryprocessor.Type
RequiredQueryIDs []uuid.UUID
Version int32
Config interface{}
}
type ListFilters struct {
Types []queryprocessor.Type
}
type Test struct {
ID uuid.UUID
QueryID uuid.UUID
DocumentID uuid.UUID
QueryVersion int32
}
type Service struct {
@@ -33,27 +27,11 @@ func New(db *repository.Queries) *Service {
return &Service{db}
}
func (s *Service) Get(ctx context.Context, id uuid.UUID) (*Query, error) {
func (s *Service) Get(ctx context.Context, id uuid.UUID) (*queryprocessor.Query, error) {
query, err := s.db.GetQuery(ctx, database.MustToDBUUID(id))
if err != nil {
return nil, err
}
queryType, err := queryprocessor.ParseDBType(query.Type)
if err != nil {
return nil, err
}
reqIDs := make([]uuid.UUID, len(query.Requiredids))
for index, id := range query.Requiredids {
reqIDs[index] = database.MustToUUID(id)
}
return &Query{
ID: id,
Type: queryType,
Version: query.Activeversion,
RequiredQueryIDs: reqIDs,
Config: string(query.Config),
}, nil
return queryprocessor.ParseDBGetQuery(&query)
}
internal/query/test.go
+2 -2
View File
@@ -6,8 +6,8 @@ import (
func (s *Service) Test(ctx context.Context, filters Test) (string, error) {
// TODO
// Sync doc
// Run test
// Sync doc - documentID
// Run test - queryID, queryVersion
return "", nil
}
internal/queryProcessor/parse.go
+41 -1
View File
@@ -42,7 +42,7 @@ func ToDBQueryType(t Type) (repository.NullQuerytype, error) {
return repository.NullQuerytype{Querytype: dbType, Valid: true}, nil
}
func ParseDBQuery(q *repository.GetCollectorQueriesRow) (*Query, error) {
func ParseDBCollectorQuery(q *repository.GetCollectorQueriesRow) (*Query, error) {
reqQueryIDs := make([]uuid.UUID, len(q.Requiredids))
for index, id := range q.Requiredids {
reqQueryIDs[index] = database.MustToUUID(id)
@@ -60,3 +60,43 @@ func ParseDBQuery(q *repository.GetCollectorQueriesRow) (*Query, error) {
RequiredQueryIDs: reqQueryIDs,
}, nil
}
func ParseDBListQuery(q *repository.ListQueriesRow) (*Query, error) {
reqQueryIDs := make([]uuid.UUID, len(q.Requiredids))
for index, id := range q.Requiredids {
reqQueryIDs[index] = database.MustToUUID(id)
}
qType, err := ParseDBType(q.Type)
if err != nil {
return nil, err
}
return &Query{
ID: database.MustToUUID(q.ID),
Version: q.Activeversion,
Type: qType,
RequiredQueryIDs: reqQueryIDs,
Config: string(q.Config),
}, nil
}
func ParseDBGetQuery(q *repository.GetQueryRow) (*Query, error) {
reqQueryIDs := make([]uuid.UUID, len(q.Requiredids))
for index, id := range q.Requiredids {
reqQueryIDs[index] = database.MustToUUID(id)
}
qType, err := ParseDBType(q.Type)
if err != nil {
return nil, err
}
return &Query{
ID: database.MustToUUID(q.ID),
Version: q.Activeversion,
Type: qType,
RequiredQueryIDs: reqQueryIDs,
Config: string(q.Config),
}, nil
}
internal/queryProcessor/service.go
+3 -3
View File
@@ -17,13 +17,13 @@ const (
type Create struct {
Type Type
RequiredQueryIDs []uuid.UUID
Config interface{}
Config string
}
type Update struct {
ID uuid.UUID
RequiredQueryIDs []uuid.UUID
Config interface{}
Config string
}
type Query struct {
@@ -31,7 +31,7 @@ type Query struct {
Type Type
Version int32
RequiredQueryIDs []uuid.UUID
Config interface{}
Config string
}
type Creator interface {
internal/queryQueue/create.go
+1 -1
View File
@@ -38,7 +38,7 @@ func (c *Queue) getCollectorQueries(ctx context.Context) error {
cleanQueries := make([]queryprocessor.Query, len(queries))
for index, dbQuery := range queries {
cleanQuery, err := queryprocessor.ParseDBQuery(&dbQuery)
cleanQuery, err := queryprocessor.ParseDBCollectorQuery(&dbQuery)
if err != nil {
return err
}