Skip to content

feat: separate structure and profile in tables yaml #338

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
adrienaury wants to merge 6 commits into
base: main
Choose a base branch
from
Open

feat: separate structure and profile in tables yaml #338

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4a196a3
feat(id): wip! add formats to ingress descriptor
adrienaury Dec 9, 2024
60ca068
feat(id): wip! add formats to start table
adrienaury Dec 9, 2024
c487ffc
feat(id): add formats to start table
adrienaury Dec 9, 2024
49ead49
feat(id): fix unit tests
adrienaury Dec 11, 2024
398a140
feat(id): add format unit test
adrienaury Dec 11, 2024
593b1f3
feat(id): add formats to push command
adrienaury Dec 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion internal/app/id/create.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ func newCreateCommand(fullName string, err *os.File, out *os.File, in *os.File)

reader := infra.NewRelationReader(relations)

e := id.Create(table, []string{}, reader, idStorageFactory(ingressDescriptor))
e := id.Create(table, []string{}, nil, reader, idStorageFactory(ingressDescriptor))
if e != nil {
fmt.Fprintln(err, e.Description)
os.Exit(1)
Expand Down
25 changes: 16 additions & 9 deletions internal/app/pull/cli.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
os.Exit(1)
}

plan, start, startSelect, e2 := getPullerPlan(idStorageFactory(table, ingressDescriptor))
plan, start, startSelect, formats, e2 := getPullerPlan(idStorageFactory(table, ingressDescriptor))
if e2 != nil {
fmt.Fprintln(err, e2.Error())
os.Exit(1)
Expand Down Expand Up @@ -175,7 +175,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
}

puller := pull.NewPullerParallel(plan, datasource, pullExporterFactory(out), tracer, parallel)
if e3 := puller.Pull(start, filter, startSelect, filters, filtersEx); e3 != nil {
if e3 := puller.Pull(start, filter, startSelect, formats, filters, filtersEx); e3 != nil {
log.Fatal().AnErr("error", e3).Msg("Fatal error stop the pull command")
os.Exit(1)
}
Expand Down Expand Up @@ -222,26 +222,26 @@ func getDataSource(dataconnectorName string, out io.Writer) (pull.DataSource, er
return datasourceFactory.New(u.URL.String(), alias.Schema), nil
}

func getPullerPlan(idStorage id.Storage) (pull.Plan, pull.Table, []string, error) {
func getPullerPlan(idStorage id.Storage) (pull.Plan, pull.Table, []string, map[string]string, error) {
pp, err1 := id.GetPullerPlan(idStorage)
if err1 != nil {
return pull.Plan{}, pull.Table{}, []string{}, err1
return pull.Plan{}, pull.Table{}, []string{}, nil, err1
}

relations, err2 := relStorage.List()
if err2 != nil {
return pull.Plan{}, pull.Table{}, []string{}, err2
return pull.Plan{}, pull.Table{}, []string{}, nil, err2
}

tables, err3 := tabStorage.List()
if err3 != nil {
return pull.Plan{}, pull.Table{}, []string{}, err3
return pull.Plan{}, pull.Table{}, []string{}, nil, err3
}

builder := newBuilder(pp, relations, tables)
plan, startTable, err4 := builder.plan()
if err4 != nil {
return pull.Plan{}, pull.Table{}, []string{}, err4
return pull.Plan{}, pull.Table{}, []string{}, nil, err4
}

// Check startTable existe in table.yaml
Expand All @@ -256,8 +256,15 @@ func getPullerPlan(idStorage id.Storage) (pull.Plan, pull.Table, []string, error

if !tableExiste {
err5 := fmt.Errorf("Table '%s' does not exist in table.yaml", string(startTable.Name))
return pull.Plan{}, pull.Table{}, []string{}, err5
return pull.Plan{}, pull.Table{}, []string{}, nil, err5
}

return plan, startTable, pp.Select(), nil
formats := map[string]string{}
if pp.Formats() != nil {
for _, column := range pp.Formats().Columns() {
formats[column] = pp.Formats().Get(column).Export()
}
}

return plan, startTable, pp.Select(), formats, nil
}
4 changes: 2 additions & 2 deletions internal/app/pull/http.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,7 @@ func HandlerFactory(ingressDescriptor string) func(w http.ResponseWriter, r *htt
return
}

plan, start, startSelect, e2 := getPullerPlan(idStorageFactory(query.Get("table"), ingressDescriptor))
plan, start, startSelect, formats, e2 := getPullerPlan(idStorageFactory(query.Get("table"), ingressDescriptor))
if e2 != nil {
log.Error().Err(e2).Msg("")
w.WriteHeader(http.StatusInternalServerError)
Expand All @@ -141,7 +141,7 @@ func HandlerFactory(ingressDescriptor string) func(w http.ResponseWriter, r *htt
pullExporter := pullExporterFactory(w)
puller := pull.NewPuller(plan, datasource, pullExporter, pull.NoTraceListener{})

e3 := puller.Pull(start, pull.Filter{Limit: limit, Values: pull.Row{}, Where: where, Distinct: distinct}, startSelect, nil, nil)
e3 := puller.Pull(start, pull.Filter{Limit: limit, Values: pull.Row{}, Where: where, Distinct: distinct}, startSelect, formats, nil, nil)
if e3 != nil {
log.Error().Err(e3).Msg("")
w.WriteHeader(http.StatusInternalServerError)
Expand Down
21 changes: 14 additions & 7 deletions internal/app/push/cli.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
os.Exit(1)
}

plan, e2 := getPlan(idStorageFactory(table, ingressDescriptor), autoTruncate)
plan, formats, e2 := getPlan(idStorageFactory(table, ingressDescriptor), autoTruncate)
if e2 != nil {
fmt.Fprintln(err, e2.Error())
os.Exit(2)
Expand Down Expand Up @@ -161,7 +161,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
observers = append(observers, observer)
}

e3 := push.Push(rowIteratorFactory(in), datadestination, plan, mode, commitSize, disableConstraints, rowExporter, translator, whereField, savepoint, autoTruncate, observers...)
e3 := push.Push(rowIteratorFactory(in), datadestination, plan, mode, commitSize, disableConstraints, rowExporter, translator, whereField, savepoint, autoTruncate, formats, observers...)
if e3 != nil {
log.Fatal().AnErr("error", e3).Msg("Fatal error stop the push command")
os.Exit(1)
Expand Down Expand Up @@ -253,20 +253,20 @@ func getDataDestination(dataconnectorName string) (push.DataDestination, *push.E
return datadestinationFactory.New(u.URL.String(), alias.Schema), nil
}

func getPlan(idStorage id.Storage, autoTruncate bool) (push.Plan, *push.Error) {
func getPlan(idStorage id.Storage, autoTruncate bool) (push.Plan, map[string]string, *push.Error) {
id, err1 := idStorage.Read()
if err1 != nil {
return nil, &push.Error{Description: err1.Error()}
return nil, nil, &push.Error{Description: err1.Error()}
}

relations, err2 := relStorage.List()
if err2 != nil {
return nil, &push.Error{Description: err2.Error()}
return nil, nil, &push.Error{Description: err2.Error()}
}

tables, err3 := tabStorage.List()
if err3 != nil {
return nil, &push.Error{Description: err3.Error()}
return nil, nil, &push.Error{Description: err3.Error()}
}

rmap := map[string]relation.Relation{}
Expand All @@ -286,7 +286,14 @@ func getPlan(idStorage id.Storage, autoTruncate bool) (push.Plan, *push.Error) {
pushtmap: map[string]push.Table{},
}

return converter.getPlan(id, autoTruncate), nil
formats := map[string]string{}
if id.Formats() != nil {
for _, column := range id.Formats().Columns() {
formats[column] = id.Formats().Get(column).Export()
}
}

return converter.getPlan(id, autoTruncate), formats, nil
}

type idToPushConverter struct {
Expand Down
4 changes: 2 additions & 2 deletions internal/app/push/http.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ func Handler(w http.ResponseWriter, r *http.Request, mode push.Mode, ingressDesc
}
}

plan, e2 := getPlan(idStorageFactory(query.Get("table"), ingressDescriptor), autoTruncate)
plan, formats, e2 := getPlan(idStorageFactory(query.Get("table"), ingressDescriptor), autoTruncate)
if e2 != nil {
log.Error().Err(e2).Msg("")
w.WriteHeader(http.StatusNotFound)
Expand Down Expand Up @@ -146,7 +146,7 @@ func Handler(w http.ResponseWriter, r *http.Request, mode push.Mode, ingressDesc

log.Debug().Msg(fmt.Sprintf("call Push with mode %s", mode))

e3 := push.Push(rowIteratorFactory(r.Body), datadestination, plan, mode, commitSize, disableConstraints, push.NoErrorCaptureRowWriter{}, nil, query.Get("using-pk-field"), "", false)
e3 := push.Push(rowIteratorFactory(r.Body), datadestination, plan, mode, commitSize, disableConstraints, push.NoErrorCaptureRowWriter{}, nil, query.Get("using-pk-field"), "", false, formats)
if e3 != nil {
log.Error().Err(e3).Msg("")
w.WriteHeader(http.StatusNotFound)
Expand Down
2 changes: 1 addition & 1 deletion internal/infra/id/table_storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,5 +38,5 @@ func (s *TableStorage) Store(adef id.IngressDescriptor) *id.Error {

// Read create new Ingress Descriptor with table as start table without relations
func (s *TableStorage) Read() (id.IngressDescriptor, *id.Error) {
return id.NewIngressDescriptor(s.table, []string{}, id.NewIngressRelationList([]id.IngressRelation{})), nil
return id.NewIngressDescriptor(s.table, []string{}, id.NewIngressColumnFormatList(map[string]id.IngressColumnFormat{}), id.NewIngressRelationList([]id.IngressRelation{})), nil
}
19 changes: 15 additions & 4 deletions internal/infra/id/yaml_storage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,9 +33,15 @@ type YAMLStructure struct {

// YAMLIngressDescriptor defines how to store an ingress descriptor in YAML format.
type YAMLIngressDescriptor struct {
StartTable string `yaml:"startTable"`
Select []string `yaml:"select"`
Relations []YAMLRelation `yaml:"relations"`
StartTable string `yaml:"startTable"`
Select []string `yaml:"select"`
Formats map[string]YAMLFormat `yaml:"formats"`
Relations []YAMLRelation `yaml:"relations"`
}

type YAMLFormat struct {
imp string `yaml:"import"`
exp string `yaml:"export"`
}

// YAMLRelation defines how to store a relation in YAML format.
Expand Down Expand Up @@ -100,6 +106,11 @@ func (s *YAMLStorage) Read() (id.IngressDescriptor, *id.Error) {
return nil, err
}

formats := map[string]id.IngressColumnFormat{}
for name, format := range structure.IngressDescriptor.Formats {
formats[name] = id.NewIngressColumnFormat(format.imp, format.exp)
}

relations := []id.IngressRelation{}
for _, relation := range structure.IngressDescriptor.Relations {
relations = append(relations,
Expand All @@ -116,7 +127,7 @@ func (s *YAMLStorage) Read() (id.IngressDescriptor, *id.Error) {
)
}

return id.NewIngressDescriptor(id.NewTable(structure.IngressDescriptor.StartTable), structure.IngressDescriptor.Select, id.NewIngressRelationList(relations)), nil
return id.NewIngressDescriptor(id.NewTable(structure.IngressDescriptor.StartTable), structure.IngressDescriptor.Select, id.NewIngressColumnFormatList(formats), id.NewIngressRelationList(relations)), nil
}

func writeFile(structure *YAMLStructure, filename string) *id.Error {
Expand Down
20 changes: 10 additions & 10 deletions pkg/id/driver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import (
)

// Create and store ingress descriptor for the given start table and relation set.
func Create(startTable string, selectColumns []string, relReader RelationReader, storage Storage) *Error {
func Create(startTable string, selectColumns []string, formats IngressColumnFormatList, relReader RelationReader, storage Storage) *Error {
relations, err := relReader.Read()
if err != nil {
return err
Expand Down Expand Up @@ -63,7 +63,7 @@ func Create(startTable string, selectColumns []string, relReader RelationReader,
}
}

id := NewIngressDescriptor(NewTable(startTable), selectColumns, NewIngressRelationList(adrelations))
id := NewIngressDescriptor(NewTable(startTable), selectColumns, formats, NewIngressRelationList(adrelations))

err = storage.Store(id)
if err != nil {
Expand All @@ -89,7 +89,7 @@ func SetStartTable(table Table, storage Storage) *Error {
return &Error{Description: fmt.Sprintf("Table %s doesn't exist", table.Name())}
}

updatedID := NewIngressDescriptor(table, id.Select(), id.Relations())
updatedID := NewIngressDescriptor(table, id.Select(), id.Formats(), id.Relations())

err = storage.Store(updatedID)
if err != nil {
Expand Down Expand Up @@ -119,7 +119,7 @@ func SetChildLookup(relation string, flag bool, storage Storage) *Error {
relations[i] = rel
}

updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations))
updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations))

err = storage.Store(updatedID)
if err != nil {
Expand Down Expand Up @@ -149,7 +149,7 @@ func SetParentLookup(relation string, flag bool, storage Storage) *Error {
relations[i] = rel
}

updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations))
updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations))

err = storage.Store(updatedID)
if err != nil {
Expand Down Expand Up @@ -179,7 +179,7 @@ func SetChildWhere(relation string, where string, storage Storage) *Error {
relations[i] = rel
}

updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations))
updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations))

err = storage.Store(updatedID)
if err != nil {
Expand Down Expand Up @@ -209,7 +209,7 @@ func SetChildSelect(relation string, columns []string, storage Storage) *Error {
relations[i] = rel
}

updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations))
updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations))

err = storage.Store(updatedID)
if err != nil {
Expand Down Expand Up @@ -239,7 +239,7 @@ func SetParentWhere(relation string, where string, storage Storage) *Error {
relations[i] = rel
}

updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations))
updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations))

err = storage.Store(updatedID)
if err != nil {
Expand Down Expand Up @@ -269,7 +269,7 @@ func SetParentSelect(relation string, columns []string, storage Storage) *Error
relations[i] = rel
}

updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations))
updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations))

err = storage.Store(updatedID)
if err != nil {
Expand Down Expand Up @@ -325,7 +325,7 @@ func GetPullerPlan(storage Storage) (PullerPlan, *Error) {
log.Warn().Msg(err.Error())
}

return NewPullerPlan(steps, g.relations, g.tables, id.Select()), nil
return NewPullerPlan(steps, g.relations, g.tables, id.Select(), id.Formats()), nil
}

// Export the puller plan.
Expand Down
Loading