ReadAttributes reads multiple attributes from the storage based on the given filter and pagination.
(ctx context.Context, tenantID string, filter *base.AttributeFilter, snap string, pagination database.Pagination)
| 350 | |
| 351 | // ReadAttributes reads multiple attributes from the storage based on the given filter and pagination. |
| 352 | func (r *DataReader) ReadAttributes(ctx context.Context, tenantID string, filter *base.AttributeFilter, snap string, pagination database.Pagination) (collection *database.AttributeCollection, ct database.EncodedContinuousToken, err error) { |
| 353 | // Start a new trace span and end it when the function exits. |
| 354 | ctx, span := internal.Tracer.Start(ctx, "data-reader.read-attributes") |
| 355 | defer span.End() |
| 356 | // Log read operation |
| 357 | slog.DebugContext(ctx, "reading attributes for tenant_id", slog.String("tenant_id", tenantID)) |
| 358 | // Decode snapshot token |
| 359 | // Decode the snapshot value. |
| 360 | var st token.SnapToken |
| 361 | st, err = snapshot.EncodedToken{Value: snap}.Decode() |
| 362 | if err != nil { |
| 363 | return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INTERNAL) |
| 364 | } |
| 365 | // Build SQL query |
| 366 | // Build the relationships query based on the provided filter, snapshot value, and pagination settings. |
| 367 | builder := r.database.Builder.Select("id, entity_type, entity_id, attribute, value").From(AttributesTable).Where(squirrel.Eq{"tenant_id": tenantID}) |
| 368 | builder = utils.AttributesFilterQueryForSelectBuilder(builder, filter) |
| 369 | builder = utils.SnapshotQuery(builder, st.(snapshot.Token).Value.Uint, st.(snapshot.Token).Snapshot) |
| 370 | |
| 371 | // Apply the pagination token and limit to the query. |
| 372 | if pagination.Token() != "" { |
| 373 | var t database.ContinuousToken |
| 374 | t, err = utils.EncodedContinuousToken{Value: pagination.Token()}.Decode() |
| 375 | if err != nil { |
| 376 | return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INVALID_CONTINUOUS_TOKEN) |
| 377 | } |
| 378 | var v uint64 |
| 379 | v, err = strconv.ParseUint(t.(utils.ContinuousToken).Value, 10, 64) |
| 380 | if err != nil { |
| 381 | return nil, nil, utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_INVALID_CONTINUOUS_TOKEN) |
| 382 | } |
| 383 | builder = builder.Where(squirrel.GtOrEq{"id": v}) |
| 384 | } |
| 385 | |
| 386 | builder = builder.OrderBy("id") |
| 387 | |
| 388 | if pagination.PageSize() != 0 { |
| 389 | builder = builder.Limit(uint64(pagination.PageSize() + 1)) |
| 390 | } |
| 391 | |
| 392 | // Generate the SQL query and arguments. |
| 393 | var query string |
| 394 | var args []interface{} |
| 395 | query, args, err = builder.ToSql() |
| 396 | if err != nil { |
| 397 | return nil, database.NewNoopContinuousToken().Encode(), utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_SQL_BUILDER) |
| 398 | } |
| 399 | // Log generated query |
| 400 | slog.DebugContext(ctx, "generated sql query", slog.String("query", query), "with args", slog.Any("arguments", args)) |
| 401 | // Execute query |
| 402 | // Execute the query and retrieve the rows. |
| 403 | var rows pgx.Rows |
| 404 | rows, err = r.database.ReadPool.Query(ctx, query, args...) |
| 405 | if err != nil { |
| 406 | return nil, database.NewNoopContinuousToken().Encode(), utils.HandleError(ctx, span, err, base.ErrorCode_ERROR_CODE_EXECUTION) |
| 407 | } |
| 408 | defer rows.Close() |
| 409 |
nothing calls this directly
no test coverage detected