Merged in feature/import (pull request #157)
Feature/import * importstart * pp * tests * importtests * 100GB * lint * passtests * utc * awsprofile * doublequotes * host
This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
package documentupload
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
"queryorchestration/internal/database/repository"
|
||||
"queryorchestration/internal/serviceconfig/objectstore"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
)
|
||||
|
||||
type File struct {
|
||||
ClientID string
|
||||
Content io.Reader
|
||||
}
|
||||
|
||||
func (s *Service) Upload(ctx context.Context, file File) error {
|
||||
return s.cfg.ExecuteDBTransaction(ctx, func(ctx context.Context, q *repository.Queries) error {
|
||||
now := time.Now().UTC()
|
||||
part, err := s.cfg.GetDBQueries().GetDocumentUploadCurrentPart(ctx, &repository.GetDocumentUploadCurrentPartParams{
|
||||
Clientid: &file.ClientID,
|
||||
Querydate: pgtype.Timestamptz{
|
||||
Valid: true,
|
||||
Time: now,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
uploadId := uuid.New()
|
||||
newPart := s.cfg.GetDirectoryPart(part.Part, part.Count)
|
||||
key := objectstore.BucketKey{
|
||||
ClientID: file.ClientID,
|
||||
Location: objectstore.Import,
|
||||
CreatedAt: now,
|
||||
EntityID: uploadId,
|
||||
Part: &newPart,
|
||||
}
|
||||
keyStr := key.String()
|
||||
bucket := s.cfg.GetBucket()
|
||||
|
||||
err = q.AddDocumentUpload(ctx, &repository.AddDocumentUploadParams{
|
||||
ID: uploadId,
|
||||
Clientid: file.ClientID,
|
||||
Bucket: bucket,
|
||||
Key: keyStr,
|
||||
Part: newPart,
|
||||
Createdat: pgtype.Timestamp{
|
||||
Valid: true,
|
||||
Time: now,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = s.cfg.GetStoreClient().PutObject(ctx, &s3.PutObjectInput{
|
||||
Bucket: &bucket,
|
||||
Key: &keyStr,
|
||||
Body: file.Content,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package documentupload_test
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"queryorchestration/internal/database/repository"
|
||||
"queryorchestration/internal/serviceconfig"
|
||||
"queryorchestration/internal/serviceconfig/objectstore"
|
||||
"queryorchestration/internal/test"
|
||||
|
||||
documentupload "queryorchestration/internal/document/upload"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
serviceconfig.BaseConfig
|
||||
objectstore.ObjectStoreConfig
|
||||
}
|
||||
|
||||
func TestUpload(t *testing.T) {
|
||||
cfg := &Config{}
|
||||
test.CreateDB(t, cfg)
|
||||
acfg := test.CreateAWSContainer(t, cfg)
|
||||
|
||||
test.SetStoreClient(t, t.Context(), cfg, acfg.ExternalEndpoint)
|
||||
test.CreateBucket(t, cfg)
|
||||
|
||||
err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{
|
||||
Clientid: "client_id",
|
||||
Name: "client",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
svc := documentupload.New(cfg)
|
||||
|
||||
file := documentupload.File{
|
||||
ClientID: "client_id",
|
||||
Content: strings.NewReader("aaaaaaaaaaaa"),
|
||||
}
|
||||
|
||||
err = svc.Upload(t.Context(), file)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package documentupload
|
||||
|
||||
import (
|
||||
"queryorchestration/internal/serviceconfig"
|
||||
"queryorchestration/internal/serviceconfig/objectstore"
|
||||
)
|
||||
|
||||
type ConfigProvider interface {
|
||||
serviceconfig.ConfigProvider
|
||||
objectstore.ConfigProvider
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
cfg ConfigProvider
|
||||
}
|
||||
|
||||
func New(cfg ConfigProvider) *Service {
|
||||
return &Service{
|
||||
cfg,
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user