pedro/meteologica
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge to one transaction for accepted and rejected data

Browse Source
This commit is contained in:
Pedro Pérez 2025-10-23 17:38:51 +02:00
parent 28c2c6a5d6
commit 062c7f3f39
1 changed files with 55 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
service_a/internal/domains/meteo/repository.go
View File

@ -9,8 +9,7 @@ import (
) )
type Repository interface { type Repository interface {
InsertAcceptedMeteoData(ctx context.Context, data []MeteoData) (int, error) InsertMeteoDataTX(ctx context.Context, accepted []MeteoData, rejected []RejectedMeteoData) (int, int, error)
InsertRejectedMeteoData(ctx context.Context, data []RejectedMeteoData) (int, error)
} }
type pgxRepo struct { type pgxRepo struct {
@ -25,13 +24,35 @@ func NewPGXRepo(pool *pgxpool.Pool) Repository {
const insertAcceptedMeteoData = `insert into public.meteo_data (location_name, max_temp, min_temp, rainfall, cloudiness, created_at) values ($1, $2, $3, $4, $5, $6) returning id` const insertAcceptedMeteoData = `insert into public.meteo_data (location_name, max_temp, min_temp, rainfall, cloudiness, created_at) values ($1, $2, $3, $4, $5, $6) returning id`
func (pgx *pgxRepo) InsertAcceptedMeteoData(ctx context.Context, data []MeteoData) (int, error) { func (pgx *pgxRepo) InsertMeteoDataTX(ctx context.Context, accepted []MeteoData, rejected []RejectedMeteoData) (int, int, error) {
tx, err := pgx.Begin(ctx) tx, err := pgx.Begin(ctx)
if err != nil { if err != nil {
return 0, fmt.Errorf("error starting transaction: %w", err) return 0, 0, fmt.Errorf("error starting transaction: %w", err)
} }
defer tx.Rollback(ctx) defer tx.Rollback(ctx)
acceptedCount, err := pgx.insertAcceptedMeteoData(ctx, tx, accepted)
if err != nil {
return 0, 0, err
}
rejectedCount, err := pgx.insertRejectedMeteoData(ctx, tx, rejected)
if err != nil {
return 0, 0, err
}
if err = tx.Commit(ctx); err != nil {
return 0, 0, fmt.Errorf("error committing transaction: %w", err)
}
return acceptedCount, rejectedCount, nil
}
func (pgx *pgxRepo) insertAcceptedMeteoData(ctx context.Context, tx b.Tx, data []MeteoData) (int, error) {
if len(data) == 0 {
return 0, nil
}
batch := &b.Batch{} batch := &b.Batch{}
for _, d := range data { for _, d := range data {
@ -40,8 +61,10 @@ func (pgx *pgxRepo) InsertAcceptedMeteoData(ctx context.Context, data []MeteoDat
results := tx.SendBatch(ctx, batch) results := tx.SendBatch(ctx, batch)
var rowsInserted int
for i := range data { for i := range data {
_, err := results.Exec() _, err := results.Exec()
rowsInserted++
if err != nil { if err != nil {
results.Close() results.Close()
return 0, fmt.Errorf("error executing batch command %d: %w", i, err) return 0, fmt.Errorf("error executing batch command %d: %w", i, err)
@ -50,13 +73,35 @@ func (pgx *pgxRepo) InsertAcceptedMeteoData(ctx context.Context, data []MeteoDat
results.Close() results.Close()
if err = tx.Commit(ctx); err != nil { return rowsInserted, nil
return 0, fmt.Errorf("error committing transaction: %w", err)
} }
const insertRejectedMeteoData = `insert into public.rejected_data (raw_data, reason) values ($1, $2) returning id`
func (pgx *pgxRepo) insertRejectedMeteoData(ctx context.Context, tx b.Tx, data []RejectedMeteoData) (int, error) {
if len(data) == 0 {
return 0, nil return 0, nil
} }
func (pgx *pgxRepo) InsertRejectedMeteoData(ctx context.Context, data []RejectedMeteoData) (int, error) { batch := &b.Batch{}
return 0, nil
for _, d := range data {
batch.Queue(insertRejectedMeteoData, d.RowValue, d.Reason)
}
results := tx.SendBatch(ctx, batch)
var rowsInserted int
for i := range data {
_, err := results.Exec()
rowsInserted++
if err != nil {
results.Close()
return 0, fmt.Errorf("error executing batch command %d: %w", i, err)
}
}
results.Close()
return rowsInserted, nil
} }