protodb

README

Alis Build ProtoDB Package (protodb)

The protodb package provides generic interfaces and utilities for standardizing database operations involving Protocol Buffer (proto.Message) resources and Google Cloud IAM policies. It simplifies building resource-oriented APIs by providing common abstractions for CRUD operations, batch processing, pagination, streaming, and error handling.

Key Concepts

ResourceTable[R proto.Message]

The ResourceTable interface defines the standard contract for a database table storing protobuf resources. It includes methods for:

• Standard CRUD: Create, Read, Write, Delete
• Batch Operations: BatchCreate, BatchRead, BatchWrite, BatchDelete
• IAM Policies: WritePolicy, BatchWritePolicies
• Query & Pagination: List, Query with pageToken and filter support.
• Streaming: Stream to retrieve resources continuously via a channel-backed iterator.

ResourceRow[R proto.Message]

The ResourceRow interface represents a single row within a database. It binds a protobuf resource to its row key and IAM policy, and provides built-in methods for applying updates:

• Data Access: GetRowKey, GetResource, GetPolicy
• Merge: Merges an updated message into the resource based on provided field mask paths.
• ApplyReadMask: Applies a fieldmaskpb.FieldMask to the resource, filtering out unrequested fields.
• Update / Delete: Directly executes modifications for the specific row in the database.

StreamResponse[T]

A generic streaming iterator returned by ResourceTable.Stream(). It manages asynchronous item retrieval using Go channels and a sync.WaitGroup.

Streaming Example Usage

streamResponse, err := db.Stream(ctx, parent, 100, "", "status = 'ACTIVE'", "")
if err != nil {
    // Handle error starting the stream
}

for {
    row, err := streamResponse.Next()
    if err == io.EOF {
        break // End of stream
    }
    if err != nil {
        // Handle stream error
        break
    }

    // Process the resource
    resource := row.GetResource()
    _ = resource
}

Error Handling Utilities

The package provides helpers to easily translate underlying database errors (like Spanner or GORM errors) into standard gRPC status errors, ensuring API consistency:

• SpannerErrorToStatus(err error) error: Converts Google Cloud Spanner and Google API errors into gRPC status errors.
• GormErrorToStatus(err error) error: Converts GORM sentinel errors (e.g., gorm.ErrRecordNotFound) and underlying Spanner errors into standard gRPC status codes (like codes.NotFound or codes.InvalidArgument).

Documentation

Overview

Package protodb provides generic interfaces and utilities for standardizing database operations involving Protocol Buffer resources and IAM policies.

Index

• func GormErrorToStatus(err error) error
• func SpannerErrorToStatus(err error) error
• type ResourceRow
• type ResourceTable
• type StreamResponse
  • func NewStreamResponse[T interface{}]() *StreamResponse[T]
  • func (r *StreamResponse[T]) AddItem(item T)
  • func (r *StreamResponse[T]) Close()
  • func (r *StreamResponse[T]) Next() (T, error)
  • func (r *StreamResponse[T]) SetError(err error)
  • func (r *StreamResponse[T]) Wait()

Functions

func GormErrorToStatus

func GormErrorToStatus(err error) error

GormErrorToStatus converts GORM and underlying Spanner errors to a gRPC status error.

func SpannerErrorToStatus

func SpannerErrorToStatus(err error) error

SpannerErrorToStatus converts a Spanner error to a gRPC status error.

Types

type ResourceRow

type ResourceRow[R proto.Message] interface {
	// GetRowKey returns the key of the row.
	GetRowKey() string
	// SetRowKey sets the key of the row.
	SetRowKey(key string)
	// GetResource returns the resource associated with the row.
	GetResource() R
	// SetResource sets the resource associated with the row.
	SetResource(resource R)
	// GetPolicy returns the IAM policy associated with the row.
	GetPolicy() *iampb.Policy
	// Merge merges the updatedMsg into the resource. The fieldMaskPaths are the paths of the fields to update.
	Merge(updatedMsg proto.Message, paths ...string)
	// ApplyReadMask applies a field mask to the resource, filtering out fields that are not in the mask.
	// ignoredPaths are paths that should be ignored when applying the read mask. These paths will always be included in the resource.
	ApplyReadMask(readMask *fieldmaskpb.FieldMask, ignoredPaths ...string) error
	// Update updates the resource in the database.
	Update(ctx context.Context) error
	// Delete deletes the resource from the database.
	Delete(ctx context.Context) error
}

ResourceRow is an interface that represents a row in the database that contains a resource and its associated IAM policy. It provides methods to get and set the row key, get the resource, get the policy, merge an updated message into the resource, apply a read mask to the resource, and update the resource in the database.

type ResourceTable

type ResourceTable[R proto.Message] interface {
	// WritePolicy writes the IAM policy for a resource.
	WritePolicy(ctx context.Context, name string, policy *iampb.Policy) error
	// BatchWritePolicies writes IAM policies for multiple resources.
	BatchWritePolicies(ctx context.Context, names []string, policies []*iampb.Policy) error

	// Create creates a new resource in the database with the given name and resource.
	// It also accepts an IAM policy to be associated with the resource.
	// It fails if a resource with the same name already exists.
	Create(ctx context.Context, name string, resource R, policy *iampb.Policy) (row ResourceRow[R], err error)
	// BatchCreate creates multiple resources in the database with the given names and resources.
	// It also accepts a list of IAM policies to be associated with each resource.
	// It fails if a resource with the same name already exists.
	BatchCreate(ctx context.Context, names []string, resources []R, policies []*iampb.Policy) (rows []ResourceRow[R], err error)
	// Read retrieves a resource by its name from the database.
	// It returns an error if the resource does not exist.
	Read(ctx context.Context, name string) (row ResourceRow[R], err error)
	// BatchRead retrieves multiple resources by their names from the database.
	// It returns a slice of ResourceRow and a slice of names that were not found.
	BatchRead(ctx context.Context, names []string) (row []ResourceRow[R], notFound []string, err error)
	// Write creates or updates resource in the database with the given name and resource.
	// It also accepts an IAM policy to be associated with the resource.
	// If the resource already exists, it updates the resource and the policy.
	Write(ctx context.Context, name string, resource R, policy *iampb.Policy) (row ResourceRow[R], err error)
	// BatchWrite creates or updates multiple resources in the database with the given names and resources.
	// It also accepts a list of IAM policies to be associated with each resource.
	// If a resource already exists, it updates the resource and the policy.
	BatchWrite(ctx context.Context, names []string, resources []R, policies []*iampb.Policy) (rows []ResourceRow[R], err error)
	// List retrieves resources from the database, optionally filtered by a filter string.
	// It returns a slice of ResourceRow and a nextPageToken for pagination.
	List(ctx context.Context, parent string, pageSize int32, pageToken string, filter string, orderBy string) (rows []ResourceRow[R], nextPageToken string, err error)
	// Stream streams resources from the database, optionally filtered by a filter string
	//
	// The function returns a StreamResponse that can be used to get the items from the stream
	// Call Next() on the StreamResponse to get the next item from the stream
	// Example:
	//  streamResponse := db.Stream(ctx, parent, filter, nil)
	//  for {
	//    resource, err := streamResponse.Next()
	//    if err == io.EOF {
	//      break
	//    }
	//    if err != nil {
	//      return err
	//    }
	//    // Do something with the resource
	//  }
	Stream(ctx context.Context, parent string, pageSize int32, pageToken string, filter string, orderBy string) (responseIterator *StreamResponse[ResourceRow[R]], err error)
	// Query retrieves resources from the database, optionally filtered by a filter string.
	// It returns a slice of ResourceRow and a nextPageToken for pagination.
	Query(ctx context.Context, pageSize int32, pageToken string, filter string, orderBy string) (rows []ResourceRow[R], nextPageToken string, err error)
	// Delete deletes a resource from the database by its name.
	Delete(ctx context.Context, name string) (err error)
	// BatchDelete deletes multiple resources from the database by their names.
	BatchDelete(ctx context.Context, names []string) (err error)
}

ResourceTable is an interface that a database storing resources must implement. It provides methods to read, update, create, list, and batch operations on resources. The type parameter R is a proto.Message that represents the resource type.

type StreamResponse

type StreamResponse[T interface{}] struct {
	// contains filtered or unexported fields
}

StreamResponse is a response for a stream Call Next to get the next item from the stream

func NewStreamResponse

func NewStreamResponse[T interface{}]() *StreamResponse[T]

NewStreamResponse creates a new StreamResponse

func (*StreamResponse[T]) AddItem

func (r *StreamResponse[T]) AddItem(item T)

func (*StreamResponse[T]) Close

func (r *StreamResponse[T]) Close()

func (*StreamResponse[T]) Next

func (r *StreamResponse[T]) Next() (T, error)

Next gets the next item from the stream. It returns io.EOF when the stream is closed.

func (*StreamResponse[T]) SetError

func (r *StreamResponse[T]) SetError(err error)

func (*StreamResponse[T]) Wait

func (r *StreamResponse[T]) Wait()