Add record importing from json & csv
This commit is contained in:
Tomaž Jerman
@@ -0,0 +1,145 @@
|
||||
package decoder
|
||||
|
||||
import (
|
||||
"io"
|
||||
)
|
||||
|
||||
type (
|
||||
multiple uint
|
||||
|
||||
FlatReader interface {
|
||||
Read() ([]string, error)
|
||||
}
|
||||
|
||||
StructuredDecoder interface {
|
||||
Decode(interface{}) error
|
||||
More() bool
|
||||
}
|
||||
|
||||
flatReader struct {
|
||||
r FlatReader
|
||||
header []string
|
||||
more bool
|
||||
}
|
||||
|
||||
structuredDecoder struct {
|
||||
header []string
|
||||
d StructuredDecoder
|
||||
buf []map[string]interface{}
|
||||
}
|
||||
|
||||
// callbacks
|
||||
sdCallback func(map[string]interface{}) error
|
||||
fdCallback func([]string) error
|
||||
)
|
||||
|
||||
// flat reader
|
||||
func NewFlatReader(r FlatReader) *flatReader {
|
||||
return &flatReader{
|
||||
r: r,
|
||||
more: true,
|
||||
}
|
||||
}
|
||||
|
||||
func (dec *flatReader) get(fnc fdCallback) error {
|
||||
v, err := dec.r.Read()
|
||||
if err == io.EOF {
|
||||
dec.more = false
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return fnc(v)
|
||||
}
|
||||
|
||||
func (dec *flatReader) walk(fnc fdCallback) error {
|
||||
for dec.more {
|
||||
if err := dec.get(fnc); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (dec *flatReader) Header() []string {
|
||||
if len(dec.header) > 0 {
|
||||
return dec.header
|
||||
}
|
||||
|
||||
dec.get(func(rtr []string) error {
|
||||
dec.header = rtr
|
||||
return nil
|
||||
})
|
||||
|
||||
return dec.header
|
||||
}
|
||||
|
||||
// structured decoder
|
||||
func NewStructuredDecoder(d StructuredDecoder) *structuredDecoder {
|
||||
return &structuredDecoder{
|
||||
d: d,
|
||||
}
|
||||
}
|
||||
|
||||
func (dec *structuredDecoder) get(fnc sdCallback) error {
|
||||
if !dec.d.More() {
|
||||
return nil
|
||||
}
|
||||
|
||||
var tmp map[string]interface{}
|
||||
err := dec.d.Decode(&tmp)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return fnc(tmp)
|
||||
}
|
||||
|
||||
func (dec *structuredDecoder) exhaustBuffer(fnc sdCallback) error {
|
||||
if dec.buf != nil {
|
||||
for _, b := range dec.buf {
|
||||
fnc(b)
|
||||
}
|
||||
dec.buf = nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (dec *structuredDecoder) walk(fnc sdCallback) error {
|
||||
if err := dec.exhaustBuffer(fnc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for dec.d.More() {
|
||||
if err := dec.get(fnc); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (dec *structuredDecoder) Header() []string {
|
||||
if len(dec.header) > 0 {
|
||||
return dec.header
|
||||
}
|
||||
|
||||
var tmp []string
|
||||
dec.get(func(rtr map[string]interface{}) error {
|
||||
// buffer first row or else it will be lost
|
||||
dec.buf = append(dec.buf, rtr)
|
||||
|
||||
tmp = make([]string, len(rtr))
|
||||
i := 0
|
||||
for k := range rtr {
|
||||
tmp[i] = k
|
||||
i++
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
dec.header = tmp
|
||||
return tmp
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
package decoder
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
)
|
||||
|
||||
type (
|
||||
RecordCreator func(mod *types.Record) error
|
||||
)
|
||||
|
||||
func fmtTime(tp string) (time.Time, error) {
|
||||
return time.Parse(time.RFC3339, tp)
|
||||
}
|
||||
func fmtTimePtr(tp string) (*time.Time, error) {
|
||||
t, err := fmtTime(tp)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &t, nil
|
||||
}
|
||||
|
||||
func mapify(header []string, values []string) map[string]string {
|
||||
rtr := make(map[string]string)
|
||||
for i, v := range values {
|
||||
rtr[header[i]] = v
|
||||
}
|
||||
|
||||
return rtr
|
||||
}
|
||||
|
||||
func setSystemField(r *types.Record, name, value string) (is bool, err error) {
|
||||
switch name {
|
||||
case "recordID", "ID":
|
||||
r.ID, err = strconv.ParseUint(value, 10, 64)
|
||||
case "moduleID":
|
||||
r.ModuleID, err = strconv.ParseUint(value, 10, 64)
|
||||
case "namespaceID":
|
||||
r.NamespaceID, err = strconv.ParseUint(value, 10, 64)
|
||||
case "ownedBy":
|
||||
r.OwnedBy, err = strconv.ParseUint(value, 10, 64)
|
||||
case "createdBy":
|
||||
r.CreatedBy, err = strconv.ParseUint(value, 10, 64)
|
||||
case "createdAt":
|
||||
r.CreatedAt, err = fmtTime(value)
|
||||
case "updatedBy":
|
||||
r.UpdatedBy, err = strconv.ParseUint(value, 10, 64)
|
||||
case "updatedAt":
|
||||
r.UpdatedAt, err = fmtTimePtr(value)
|
||||
case "deletedBy":
|
||||
r.DeletedBy, err = strconv.ParseUint(value, 10, 64)
|
||||
case "deletedAt":
|
||||
r.DeletedAt, err = fmtTimePtr(value)
|
||||
default:
|
||||
return false, err
|
||||
}
|
||||
return true, err
|
||||
}
|
||||
|
||||
func (dec flatReader) Records(fields map[string]string, Create RecordCreator) error {
|
||||
header := dec.Header()
|
||||
|
||||
err := dec.walk(func(row []string) error {
|
||||
mapped := mapify(header, row)
|
||||
r := types.Record{}
|
||||
rvs := types.RecordValueSet{}
|
||||
|
||||
i := 0
|
||||
for imp, rec := range fields {
|
||||
if rec == "" {
|
||||
return errors.New("Can not import record: Record field not defined")
|
||||
}
|
||||
|
||||
val := mapped[imp]
|
||||
if system, err := setSystemField(&r, rec, val); err != nil {
|
||||
return err
|
||||
} else if !system {
|
||||
rv := types.RecordValue{
|
||||
Name: rec,
|
||||
Value: val,
|
||||
Place: uint(i),
|
||||
}
|
||||
i++
|
||||
|
||||
rvs = append(rvs, &rv)
|
||||
}
|
||||
}
|
||||
|
||||
r.Values = rvs
|
||||
return Create(&r)
|
||||
})
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (dec structuredDecoder) Records(fields map[string]string, Create RecordCreator) error {
|
||||
err := dec.walk(func(entry map[string]interface{}) error {
|
||||
r := types.Record{}
|
||||
rvs := types.RecordValueSet{}
|
||||
|
||||
i := 0
|
||||
for imp, rec := range fields {
|
||||
if rec == "" {
|
||||
return errors.New("Can not import record: Record field not defined")
|
||||
}
|
||||
|
||||
val := fmt.Sprintf("%v", entry[imp])
|
||||
if system, err := setSystemField(&r, rec, val); err != nil {
|
||||
return err
|
||||
} else if !system {
|
||||
rv := types.RecordValue{
|
||||
Name: rec,
|
||||
Value: val,
|
||||
Place: uint(i),
|
||||
}
|
||||
i++
|
||||
|
||||
rvs = append(rvs, &rv)
|
||||
}
|
||||
}
|
||||
|
||||
r.Values = rvs
|
||||
return Create(&r)
|
||||
})
|
||||
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/internal/auth"
|
||||
"github.com/titpetric/factory"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type (
|
||||
recordSet []*RecordImportSession
|
||||
|
||||
importSession struct {
|
||||
l sync.Mutex
|
||||
logger *zap.Logger
|
||||
|
||||
records recordSet
|
||||
}
|
||||
|
||||
ImportSessionService interface {
|
||||
FindRecordByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error)
|
||||
SetRecordByID(ctx context.Context, sessionID, namespaceID, moduleID uint64, fields map[string]string, progress *RecordImportProgress, decoder Decoder) (*RecordImportSession, error)
|
||||
DeleteRecordByID(ctx context.Context, sessionID uint64) error
|
||||
}
|
||||
)
|
||||
|
||||
func ImportSession() ImportSessionService {
|
||||
return &importSession{
|
||||
logger: DefaultLogger.Named("importSession"),
|
||||
records: recordSet{},
|
||||
}
|
||||
}
|
||||
|
||||
func (svc importSession) indexOf(userID, sessionID uint64) int {
|
||||
for i, r := range svc.records {
|
||||
if r.SessionID == sessionID && r.UserID == userID {
|
||||
return i
|
||||
}
|
||||
}
|
||||
|
||||
return -1
|
||||
}
|
||||
|
||||
func (svc *importSession) FindRecordByID(ctx context.Context, sessionID uint64) (*RecordImportSession, error) {
|
||||
svc.l.Lock()
|
||||
defer svc.l.Unlock()
|
||||
|
||||
userID := auth.GetIdentityFromContext(ctx).Identity()
|
||||
i := svc.indexOf(userID, sessionID)
|
||||
if i >= 0 {
|
||||
return svc.records[i], nil
|
||||
}
|
||||
return nil, errors.New("Can't access session: session not found")
|
||||
}
|
||||
|
||||
func (svc *importSession) SetRecordByID(ctx context.Context, sessionID, namespaceID, moduleID uint64, fields map[string]string, progress *RecordImportProgress, decoder Decoder) (*RecordImportSession, error) {
|
||||
svc.l.Lock()
|
||||
defer svc.l.Unlock()
|
||||
|
||||
userID := auth.GetIdentityFromContext(ctx).Identity()
|
||||
i := svc.indexOf(userID, sessionID)
|
||||
var ris *RecordImportSession
|
||||
|
||||
if i >= 0 {
|
||||
ris = svc.records[i]
|
||||
} else {
|
||||
ris = &RecordImportSession{
|
||||
SessionID: factory.Sonyflake.NextID(),
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
svc.records = append(svc.records, ris)
|
||||
ris.UserID = userID
|
||||
}
|
||||
ris.UpdatedAt = time.Now()
|
||||
|
||||
if namespaceID > 0 {
|
||||
ris.NamespaceID = namespaceID
|
||||
}
|
||||
if moduleID > 0 {
|
||||
ris.ModuleID = moduleID
|
||||
}
|
||||
if fields != nil {
|
||||
ris.Fields = fields
|
||||
}
|
||||
if progress != nil {
|
||||
ris.Progress = *progress
|
||||
}
|
||||
|
||||
if ris.Progress.FinishedAt != nil {
|
||||
ris.Decoder = nil
|
||||
} else if decoder != nil {
|
||||
ris.Decoder = decoder
|
||||
}
|
||||
|
||||
return ris, nil
|
||||
}
|
||||
|
||||
// https://stackoverflow.com/a/37335777
|
||||
func remove(s recordSet, i int) recordSet {
|
||||
s[len(s)-1], s[i] = s[i], s[len(s)-1]
|
||||
return s[:len(s)-1]
|
||||
}
|
||||
|
||||
func (svc *importSession) DeleteRecordByID(ctx context.Context, sessionID uint64) error {
|
||||
svc.l.Lock()
|
||||
defer svc.l.Unlock()
|
||||
|
||||
userID := auth.GetIdentityFromContext(ctx).Identity()
|
||||
i := svc.indexOf(userID, sessionID)
|
||||
|
||||
if i >= 0 {
|
||||
svc.records = remove(svc.records, i)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// @todo run this in some interval
|
||||
func (svc *importSession) clean(ctx context.Context) {
|
||||
svc.l.Lock()
|
||||
defer svc.l.Unlock()
|
||||
|
||||
for i := len(svc.records) - 1; i >= 0; i-- {
|
||||
r := svc.records[i]
|
||||
if time.Now().After(r.UpdatedAt.Add(time.Hour * 24 * 3)) {
|
||||
svc.records = remove(svc.records, i)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -10,11 +10,17 @@ import (
|
||||
"github.com/titpetric/factory"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/decoder"
|
||||
"github.com/cortezaproject/corteza-server/compose/internal/repository"
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/cortezaproject/corteza-server/internal/auth"
|
||||
)
|
||||
|
||||
const (
|
||||
IMPORT_ON_ERROR_SKIP = "SKIP"
|
||||
IMPORT_ON_ERROR_FAIL = "FAIL"
|
||||
)
|
||||
|
||||
type (
|
||||
record struct {
|
||||
db *factory.DB
|
||||
@@ -57,6 +63,7 @@ type (
|
||||
Report(namespaceID, moduleID uint64, metrics, dimensions, filter string) (interface{}, error)
|
||||
Find(filter types.RecordFilter) (set types.RecordSet, f types.RecordFilter, err error)
|
||||
Export(types.RecordFilter, Encoder) error
|
||||
Import(*RecordImportSession, ImportSessionService) error
|
||||
|
||||
Create(record *types.Record) (*types.Record, error)
|
||||
Update(record *types.Record) (*types.Record, error)
|
||||
@@ -67,6 +74,32 @@ type (
|
||||
Encoder interface {
|
||||
Record(*types.Record) error
|
||||
}
|
||||
|
||||
Decoder interface {
|
||||
Header() []string
|
||||
Records(fields map[string]string, Create decoder.RecordCreator) error
|
||||
}
|
||||
|
||||
RecordImportSession struct {
|
||||
Decoder Decoder `json:"-"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
OnError string `json:"onError"`
|
||||
SessionID uint64 `json:"sessionID,string"`
|
||||
UserID uint64 `json:"userID,string"`
|
||||
NamespaceID uint64 `json:"namespaceID,string"`
|
||||
ModuleID uint64 `json:"moduleID,string"`
|
||||
Fields map[string]string `json:"fields"`
|
||||
Progress RecordImportProgress `json:"progress"`
|
||||
}
|
||||
|
||||
RecordImportProgress struct {
|
||||
StartedAt *time.Time `json:"startedAt"`
|
||||
FinishedAt *time.Time `json:"finishedAt"`
|
||||
Completed uint64 `json:"completed"`
|
||||
Failed uint64 `json:"failed"`
|
||||
FailReason error `json:"failReason"`
|
||||
}
|
||||
)
|
||||
|
||||
func Record() RecordService {
|
||||
@@ -184,6 +217,49 @@ func (svc record) Find(filter types.RecordFilter) (set types.RecordSet, f types.
|
||||
return
|
||||
}
|
||||
|
||||
func (svc record) Import(ses *RecordImportSession, ssvc ImportSessionService) error {
|
||||
if ses.Decoder == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if ses.Progress.StartedAt != nil {
|
||||
return errors.New("Unable to start import: Import session already active")
|
||||
}
|
||||
|
||||
sa := time.Now()
|
||||
ses.Progress.StartedAt = &sa
|
||||
ssvc.SetRecordByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil)
|
||||
|
||||
return svc.db.Transaction(func() (err error) {
|
||||
err = ses.Decoder.Records(ses.Fields, func(mod *types.Record) error {
|
||||
mod.NamespaceID = ses.NamespaceID
|
||||
mod.ModuleID = ses.ModuleID
|
||||
mod.OwnedBy = ses.UserID
|
||||
|
||||
_, err := svc.Create(mod)
|
||||
if err != nil {
|
||||
ses.Progress.Failed++
|
||||
ses.Progress.FailReason = err
|
||||
|
||||
if ses.OnError == IMPORT_ON_ERROR_FAIL {
|
||||
fa := time.Now()
|
||||
ses.Progress.FinishedAt = &fa
|
||||
ssvc.SetRecordByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil)
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
ses.Progress.Completed++
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
fa := time.Now()
|
||||
ses.Progress.FinishedAt = &fa
|
||||
ssvc.SetRecordByID(svc.ctx, ses.SessionID, 0, 0, nil, &ses.Progress, nil)
|
||||
return
|
||||
})
|
||||
}
|
||||
|
||||
// Export returns all records
|
||||
//
|
||||
// @todo better value handling
|
||||
|
||||
@@ -47,14 +47,14 @@ var (
|
||||
// DefaultAutomationRunner runs automation scripts by listening to triggerManager and invoking Corredor service
|
||||
DefaultAutomationRunner automationRunner
|
||||
|
||||
DefaultNamespace NamespaceService
|
||||
DefaultRecord RecordService
|
||||
DefaultModule ModuleService
|
||||
DefaultChart ChartService
|
||||
DefaultPage PageService
|
||||
|
||||
DefaultAttachment AttachmentService
|
||||
DefaultNotification NotificationService
|
||||
DefaultNamespace NamespaceService
|
||||
DefaultImportSession ImportSessionService
|
||||
DefaultRecord RecordService
|
||||
DefaultModule ModuleService
|
||||
DefaultChart ChartService
|
||||
DefaultPage PageService
|
||||
DefaultAttachment AttachmentService
|
||||
DefaultNotification NotificationService
|
||||
|
||||
DefaultSystemUser *systemUser
|
||||
)
|
||||
@@ -130,6 +130,7 @@ func Init(ctx context.Context, log *zap.Logger, c Config) (err error) {
|
||||
)
|
||||
}
|
||||
|
||||
DefaultImportSession = ImportSession()
|
||||
DefaultRecord = Record()
|
||||
DefaultPage = Page()
|
||||
DefaultChart = Chart()
|
||||
|
||||
+108
-8
@@ -12,11 +12,13 @@ import (
|
||||
|
||||
"github.com/pkg/errors"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/compose/decoder"
|
||||
"github.com/cortezaproject/corteza-server/compose/encoder"
|
||||
"github.com/cortezaproject/corteza-server/compose/internal/repository"
|
||||
"github.com/cortezaproject/corteza-server/compose/internal/service"
|
||||
"github.com/cortezaproject/corteza-server/compose/rest/request"
|
||||
"github.com/cortezaproject/corteza-server/compose/types"
|
||||
"github.com/cortezaproject/corteza-server/pkg/mime"
|
||||
"github.com/cortezaproject/corteza-server/pkg/rh"
|
||||
)
|
||||
|
||||
@@ -36,10 +38,11 @@ type (
|
||||
}
|
||||
|
||||
Record struct {
|
||||
record service.RecordService
|
||||
module service.ModuleService
|
||||
attachment service.AttachmentService
|
||||
ac recordAccessController
|
||||
importSession service.ImportSessionService
|
||||
record service.RecordService
|
||||
module service.ModuleService
|
||||
attachment service.AttachmentService
|
||||
ac recordAccessController
|
||||
}
|
||||
|
||||
recordAccessController interface {
|
||||
@@ -50,10 +53,11 @@ type (
|
||||
|
||||
func (Record) New() *Record {
|
||||
return &Record{
|
||||
record: service.DefaultRecord,
|
||||
module: service.DefaultModule,
|
||||
attachment: service.DefaultAttachment,
|
||||
ac: service.DefaultAccessControl,
|
||||
importSession: service.DefaultImportSession,
|
||||
record: service.DefaultRecord,
|
||||
module: service.DefaultModule,
|
||||
attachment: service.DefaultAttachment,
|
||||
ac: service.DefaultAccessControl,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -167,6 +171,102 @@ func (ctrl *Record) Upload(ctx context.Context, r *request.RecordUpload) (interf
|
||||
return makeAttachmentPayload(ctx, a, err)
|
||||
}
|
||||
|
||||
func (ctrl *Record) ImportInit(ctx context.Context, r *request.RecordImportInit) (interface{}, error) {
|
||||
var (
|
||||
err error
|
||||
recordDecoder service.Decoder
|
||||
)
|
||||
|
||||
// Access control.
|
||||
if _, err = ctrl.module.With(ctx).FindByID(r.NamespaceID, r.ModuleID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
f, err := r.Upload.Open()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
_, ext, err := mime.Type(f)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if ext == "txt" {
|
||||
if is, err := mime.JsonL(f); err != nil {
|
||||
return nil, err
|
||||
} else if is {
|
||||
ext = "jsonl"
|
||||
}
|
||||
}
|
||||
|
||||
// determine decoder
|
||||
switch strings.ToLower(ext) {
|
||||
case "json", "jsonl", "ldjson", "ndjson":
|
||||
recordDecoder = decoder.NewStructuredDecoder(json.NewDecoder(f))
|
||||
|
||||
case "csv":
|
||||
recordDecoder = decoder.NewFlatReader(csv.NewReader(f))
|
||||
|
||||
default:
|
||||
return nil, errors.New(fmt.Sprintf("unsupported format (\"%s\")", ext))
|
||||
|
||||
}
|
||||
|
||||
header := recordDecoder.Header()
|
||||
hh := make(map[string]string)
|
||||
for _, h := range header {
|
||||
hh[h] = ""
|
||||
}
|
||||
|
||||
return ctrl.importSession.SetRecordByID(ctx, 0, r.NamespaceID, r.ModuleID, hh, nil, recordDecoder)
|
||||
}
|
||||
|
||||
func (ctrl *Record) ImportRun(ctx context.Context, r *request.RecordImportRun) (interface{}, error) {
|
||||
var (
|
||||
err error
|
||||
)
|
||||
|
||||
// Access control.
|
||||
if _, err = ctrl.module.With(ctx).FindByID(r.NamespaceID, r.ModuleID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Check if session ok
|
||||
ses, err := ctrl.importSession.FindRecordByID(ctx, r.SessionID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if ses.Progress.StartedAt != nil {
|
||||
return nil, errors.New("Unable to start import: Import session already active")
|
||||
}
|
||||
|
||||
ses.Fields = make(map[string]string)
|
||||
err = json.Unmarshal(r.Fields, &ses.Fields)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ses.OnError = r.OnError
|
||||
|
||||
// @todo routine
|
||||
ctrl.record.With(ctx).Import(ses, ctrl.importSession)
|
||||
|
||||
return ses, nil
|
||||
}
|
||||
|
||||
func (ctrl *Record) ImportProgress(ctx context.Context, r *request.RecordImportProgress) (interface{}, error) {
|
||||
// Get session
|
||||
ses, err := ctrl.importSession.FindRecordByID(ctx, r.SessionID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return ses, nil
|
||||
}
|
||||
|
||||
func (ctrl *Record) Export(ctx context.Context, r *request.RecordExport) (interface{}, error) {
|
||||
type (
|
||||
// ad-hoc interface for our encoder
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
package mime
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"io"
|
||||
|
||||
"github.com/gabriel-vasile/mimetype"
|
||||
)
|
||||
|
||||
func Type(file io.ReadSeeker) (mt string, ext string, err error) {
|
||||
if _, err = file.Seek(0, 0); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Make sure we rewind when we're done
|
||||
defer file.Seek(0, 0)
|
||||
return mimetype.DetectReader(file)
|
||||
}
|
||||
|
||||
func JsonL(file io.ReadSeeker) (bool, error) {
|
||||
// ExtractMimetype fails to detect json if jsonl is used
|
||||
// For now check if first rune is {
|
||||
r := bufio.NewReader(file)
|
||||
rn, _, err := r.ReadRune()
|
||||
defer file.Seek(0, 0)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
if string(rn) == "{" {
|
||||
return true, nil
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
Reference in New Issue
Block a user