infureal/gis
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(datasets): periods (start_date/end_date), dg_id column, date filters; rename observation dates

Browse Source 
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Bakhtiyar Issakhmetov
2026-09-05 13:16:07 +05:00
parent 37d4f3519b
commit 922812b765
10 changed files with 729 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
internal/repository/postgres/dataset.go
View File

@@ -27,7 +27,7 @@ func NewDatasetRepository(pool *pgxpool.Pool) *DatasetRepository {
// datasetColumns lists the dataset columns for SELECT and RETURNING. The
// geometry is exposed as GeoJSON (jsonb) rather than its raw EWKB form, and a
// bounding box array is derived for raster datasets only.
const datasetColumns = `id, category_id, name, description, unit, filename, storage_key, cog_storage_key, file_type, size_bytes, content_type, properties, meta, automated, status, attribute_columns, kato_column, year_columns, parse_error, ST_AsGeoJSON(geometry)::jsonb AS geometry,
const datasetColumns = `id, category_id, dg_id, to_char(start_date, 'YYYY-MM-DD') AS start_date, to_char(end_date, 'YYYY-MM-DD') AS end_date, name, description, unit, filename, storage_key, cog_storage_key, file_type, size_bytes, content_type, properties, meta, automated, status, attribute_columns, kato_column, year_columns, parse_error, ST_AsGeoJSON(geometry)::jsonb AS geometry,
CASE WHEN file_type = 'raster' AND geometry IS NOT NULL
THEN ARRAY[ST_XMin(geometry), ST_YMin(geometry), ST_XMax(geometry), ST_YMax(geometry)]
ELSE NULL END AS bbox,
@@ -37,7 +37,7 @@ const datasetColumns = `id, category_id, name, description, unit, filename, stor
func scanDataset(row pgx.Row) (domain.Dataset, error) {
var d domain.Dataset
err := row.Scan(
&d.ID, &d.CategoryID, &d.Name, &d.Description, &d.Unit,
&d.ID, &d.CategoryID, &d.DatasetGroupID, &d.StartDate, &d.EndDate, &d.Name, &d.Description, &d.Unit,
&d.Filename, &d.StorageKey, &d.CogStorageKey, &d.FileType, &d.SizeBytes, &d.ContentType,
&d.Properties, &d.Meta, &d.Automated, &d.Status,
&d.AttributeColumns, &d.KatoColumn, &d.YearColumns, &d.ParseError,
@@ -59,28 +59,28 @@ func nullableJSON(raw json.RawMessage) any {
// Create inserts a new dataset and returns the stored row.
func (r *DatasetRepository) Create(ctx context.Context, d domain.Dataset) (domain.Dataset, error) {
row := r.pool.QueryRow(ctx,
`INSERT INTO datasets (category_id, name, description, unit, filename, storage_key, file_type, size_bytes, content_type, properties, meta, automated, status)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
`INSERT INTO datasets (category_id, name, description, unit, filename, storage_key, file_type, size_bytes, content_type, properties, meta, automated, status, dg_id, start_date, end_date)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15::date, $16::date)
RETURNING `+datasetColumns,
d.CategoryID, d.Name, d.Description, d.Unit, d.Filename, d.StorageKey, d.FileType, d.SizeBytes, d.ContentType,
nullableJSON(d.Properties), nullableJSON(d.Meta), d.Automated, d.Status,
d.DatasetGroupID, d.StartDate, d.EndDate,
)
out, err := scanDataset(row)
return out, mapError(err)
}
// UpdateFields overwrites a dataset's editable metadata columns (category,
// name, description, unit). description and unit are nullable: a nil pointer
// clears the column. Returns domain.ErrNotFound if the dataset does not
// exist, or domain.ErrConflict if categoryID does not reference an existing
// category.
func (r *DatasetRepository) UpdateFields(ctx context.Context, id, categoryID uuid.UUID, name string, description, unit *string) (domain.Dataset, error) {
// UpdateFields overwrites a dataset's editable metadata columns. Nil pointers
// clear the nullable columns. Returns domain.ErrNotFound if the dataset does
// not exist, or domain.ErrConflict if the category/group does not exist.
func (r *DatasetRepository) UpdateFields(ctx context.Context, id uuid.UUID, f domain.DatasetFields) (domain.Dataset, error) {
row := r.pool.QueryRow(ctx,
`UPDATE datasets
SET category_id = $2, name = $3, description = $4, unit = $5, updated_at = now()
SET category_id = $2, name = $3, description = $4, unit = $5,
dg_id = $6, start_date = $7::date, end_date = $8::date, updated_at = now()
WHERE id = $1
RETURNING `+datasetColumns,
id, categoryID, name, description, unit,
id, f.CategoryID, f.Name, f.Description, f.Unit, f.DatasetGroupID, f.StartDate, f.EndDate,
)
out, err := scanDataset(row)
return out, mapError(err)
@@ -276,23 +276,23 @@ func (r *DatasetRepository) ReplaceObservations(ctx context.Context, datasetID u
if len(obs) > 0 {
rows := make([][]any, len(obs))
for i, o := range obs {
d, err := time.Parse("2006-01-02", o.Date)
d, err := time.Parse("2006-01-02", o.StartDate)
if err != nil {
return fmt.Errorf("invalid observation date %q: %w", o.Date, err)
return fmt.Errorf("invalid observation start_date %q: %w", o.StartDate, err)
}
var dateTo any
if o.DateTo != nil {
dt, err := time.Parse("2006-01-02", *o.DateTo)
var endDate any
if o.EndDate != nil {
dt, err := time.Parse("2006-01-02", *o.EndDate)
if err != nil {
return fmt.Errorf("invalid observation date_to %q: %w", *o.DateTo, err)
return fmt.Errorf("invalid observation end_date %q: %w", *o.EndDate, err)
}
dateTo = dt
endDate = dt
}
rows[i] = []any{datasetID, o.KatoCode, d, dateTo, o.Value, o.ValueText}
rows[i] = []any{datasetID, o.KatoCode, d, endDate, o.Value, o.ValueText}
}
_, err := tx.CopyFrom(ctx,
pgx.Identifier{"dataset_observations"},
[]string{"dataset_id", "kato_code", "date", "date_to", "value", "value_text"},
[]string{"dataset_id", "kato_code", "start_date", "end_date", "value", "value_text"},
pgx.CopyFromRows(rows),
)
if err != nil {
@@ -303,16 +303,16 @@ func (r *DatasetRepository) ReplaceObservations(ctx context.Context, datasetID u
return mapError(tx.Commit(ctx))
}
const observationColumns = `id, dataset_id, kato_code, to_char(date, 'YYYY-MM-DD') AS date, to_char(date_to, 'YYYY-MM-DD') AS date_to, value, value_text`
const observationColumns = `id, dataset_id, kato_code, to_char(start_date, 'YYYY-MM-DD') AS start_date, to_char(end_date, 'YYYY-MM-DD') AS end_date, value, value_text`
func scanObservation(row pgx.Row) (domain.Observation, error) {
var o domain.Observation
err := row.Scan(&o.ID, &o.DatasetID, &o.KatoCode, &o.Date, &o.DateTo, &o.Value, &o.ValueText)
err := row.Scan(&o.ID, &o.DatasetID, &o.KatoCode, &o.StartDate, &o.EndDate, &o.Value, &o.ValueText)
return o, err
}
// ListObservations returns a page of observations for a dataset, optionally
// filtered by KATO code, ordered by (kato_code, date).
// filtered by KATO code, ordered by (kato_code, start_date).
func (r *DatasetRepository) ListObservations(ctx context.Context, datasetID uuid.UUID, katoCode *string, limit, offset int) ([]domain.Observation, error) {
base := `SELECT ` + observationColumns + ` FROM dataset_observations WHERE dataset_id = $1`
@@ -322,11 +322,11 @@ func (r *DatasetRepository) ListObservations(ctx context.Context, datasetID uuid
)
if katoCode != nil {
rows, err = r.pool.Query(ctx,
base+` AND kato_code = $2 ORDER BY kato_code, date LIMIT $3 OFFSET $4`,
base+` AND kato_code = $2 ORDER BY kato_code, start_date LIMIT $3 OFFSET $4`,
datasetID, *katoCode, limit, offset)
} else {
rows, err = r.pool.Query(ctx,
base+` ORDER BY kato_code, date LIMIT $2 OFFSET $3`,
base+` ORDER BY kato_code, start_date LIMIT $2 OFFSET $3`,
datasetID, limit, offset)
}
if err != nil {
@@ -368,12 +368,25 @@ func (r *DatasetRepository) GetByID(ctx context.Context, id uuid.UUID) (domain.D
return out, mapError(err)
}
const datasetSummaryColumns = `id, category_id, name, description, unit, file_type, size_bytes, status, created_at, updated_at`
// LatestInGroup returns the freshest dataset of a group: greatest end_date
// (NULL end_date sorts last), ties broken by created_at DESC. status and
// fileType optionally restrict candidates. Returns domain.ErrNotFound when no
// dataset matches.
func (r *DatasetRepository) LatestInGroup(ctx context.Context, groupID uuid.UUID, status *string, fileType *domain.FileType) (domain.Dataset, error) {
where, args := datasetFilterClause(domain.DatasetFilter{DatasetGroupID: &groupID, Status: status, FileType: fileType}, nil)
row := r.pool.QueryRow(ctx,
`SELECT `+datasetColumns+` FROM datasets`+where+
` ORDER BY end_date DESC NULLS LAST, created_at DESC LIMIT 1`, args...)
out, err := scanDataset(row)
return out, mapError(err)
}
const datasetSummaryColumns = `id, category_id, dg_id, to_char(start_date, 'YYYY-MM-DD') AS start_date, to_char(end_date, 'YYYY-MM-DD') AS end_date, name, description, unit, file_type, size_bytes, status, created_at, updated_at`
func scanDatasetSummary(row pgx.Row) (domain.DatasetSummary, error) {
var d domain.DatasetSummary
err := row.Scan(
&d.ID, &d.CategoryID, &d.Name, &d.Description, &d.Unit,
&d.ID, &d.CategoryID, &d.DatasetGroupID, &d.StartDate, &d.EndDate, &d.Name, &d.Description, &d.Unit,
&d.FileType, &d.SizeBytes, &d.Status, &d.CreatedAt, &d.UpdatedAt,
)
return d, err
@@ -400,6 +413,22 @@ func datasetFilterClause(f domain.DatasetFilter, args []any) (string, []any) {
args = append(args, *f.Status)
conds = append(conds, fmt.Sprintf("status = $%d", len(args)))
}
if f.DatasetGroupID != nil {
args = append(args, *f.DatasetGroupID)
conds = append(conds, fmt.Sprintf("dg_id = $%d", len(args)))
}
for _, b := range []struct {
col, op string
val *string
}{
{"start_date", ">=", f.StartDateFrom}, {"start_date", "<=", f.StartDateTo},
{"end_date", ">=", f.EndDateFrom}, {"end_date", "<=", f.EndDateTo},
} {
if b.val != nil {
args = append(args, *b.val)
conds = append(conds, fmt.Sprintf("%s %s $%d::date", b.col, b.op, len(args)))
}
}
if len(conds) == 0 {
return "", args
}