Guia — IDataProvider

Um banco de dados novo, ativado por provider: num spec.dataSources. QuerySpec carrega filtros/ordenação/paginação numa forma neutra — não há obrigação de suportar SQL.

Exemplo mínimo

using Pipevine.Abstractions;
using Pipevine.Core.Configuration;
using Pipevine.Data.Abstractions;
using Pipevine.Model.Resources;

// A chave usada no YAML: `provider: acmedb`
[Provider("acmedb", ProviderKind.DataSource)]
public sealed class AcmeDataProvider : IDataProvider
{
    public IDataSession CreateSession(DataSourceDefinition definition)
    {
        var config = PipevineNodeJsonConverter.ToJsonNode(definition.Config);
        var connectionString = config?["connectionString"]?.GetValue<string>()
            ?? throw new DataProviderConfigurationException($"Data source '{definition.Id}' requer 'connectionString'.");

        return new AcmeDataSession(connectionString);
    }
}

IDataSession : IPipelineTransaction — é literalmente a transação que o executor do pipeline comita/reverte quando a rota declara transaction: required (o motor de pipeline só conhece o contrato genérico, nunca que é especificamente um banco de dados):

internal sealed class AcmeDataSession(string connectionString) : IDataSession
{
    public IEntityStore GetStore(string entityName) => new AcmeEntityStore(entityName /* ... */);

    public ValueTask<int> ExecuteSqlAsync(string sql, IReadOnlyDictionary<string, JsonNode?> parameters, CancellationToken ct) =>
        throw new NotSupportedException("acmedb não tem dialeto SQL — use os steps db.* tipados.");

    public ValueTask CommitAsync(CancellationToken ct) { /* ... */ }
    public ValueTask RollbackAsync(CancellationToken ct) { /* ... */ }
    public ValueTask DisposeAsync() { /* ... */ }
}

db.sql é o escape hatch declarado no vocabulário do YAML — um provider não relacional não precisa implementá-lo de verdade; lançar NotSupportedException é aceitável quando esse caminho realmente não existe no banco de destino.

IEntityStore é onde QuerySpec entra:

internal sealed class AcmeEntityStore(string entityName) : IEntityStore
{
    public ValueTask<JsonNode?> FindAsync(QuerySpec spec, CancellationToken ct) { /* traduza spec.Filters */ }
    public ValueTask<bool> ExistsAsync(QuerySpec spec, CancellationToken ct) { /* ... */ }
    public ValueTask<IReadOnlyList<JsonNode>> QueryAsync(QuerySpec spec, CancellationToken ct) { /* spec.Sort/Skip/Take */ }
    public ValueTask PersistAsync(JsonNode data, OnConflictSpec? onConflict, CancellationToken ct) { /* ... */ }
    public ValueTask UpdateAsync(QuerySpec spec, JsonNode patch, CancellationToken ct) { /* ... */ }
    public ValueTask DeleteAsync(QuerySpec spec, CancellationToken ct) { /* ... */ }
}

QuerySpec.Filters cobre oito operadores (Eq/Ne/Gt/Gte/Lt/Lte/Contains/In) — se o seu banco não conseguir expressar algum deles, isso é sinal de que vale revisitar como a abstração está sendo usada, não que você deva forçar um NotSupportedException silencioso num caso comum.

Ativação

spec:
  dataSources:
    - { id: main, provider: acmedb, connectionString: "${env:ACME_DB_CONNECTION}" }
  entities:
    - { name: item, table: items, dataSource: main, key: id, properties: [...] }

services.AddKeyedSingleton<IDataProvider, AcmeDataProvider>("acmedb");
// mais services.AddPipevineData() do core, que registra os steps genéricos db.persist/find/...

Se provider: acmedb aparecer no YAML sem essa chave registrada, o build falha com um diagnóstico claro:

orders.pipevine.yaml(14,20): error PV3010: nenhum provider de dados registrado
para a chave 'acmedb'. Referencie o pacote que a fornece.

Capacidades (opcional, mas declare a verdade)

IDataProvider.Capabilities tem um valor default de "tudo ligado" — não implementar a propriedade é válido e comum (todo provider relacional faz isso). Mas se o seu banco não tem alguma dessas quatro capacidades, declare:

public DataProviderCapabilities Capabilities =>
    DataProviderCapabilities.AtomicMultiWrite | DataProviderCapabilities.ReadYourOwnWrites | DataProviderCapabilities.OffsetPaging;
    // sem RawStatements: acmedb não tem dialeto SQL
  • RawStatements ausente → db.sql contra este provider é recusado em build-time, com uma mensagem nomeando o data source e sugerindo os steps db.* tipados.
  • OffsetPaging ausente → db.query com skip: é recusado em build-time.
  • AtomicMultiWrite ausente → transaction: required é recusado no startup, contra a conexão real (não no build, se sua resposta depende de handshake com o servidor — como o MongoDB, onde só um replica set suporta transação multi-documento; se sua resposta é sempre a mesma, um valor estático já é suficiente).
  • ReadYourOwnWrites ausente → um pipeline transaction: required que lê (db.find/db.exists/ db.query) uma entidade que ele mesmo escreveu antes (db.persist/db.update/db.delete) é recusado no startup, nomeando os dois steps envolvidos. Isso é o caso de uma sessão que bufferiza escritas até o commit em vez de aplicá-las na hora (Cassandra/DynamoDB: acumulam num buffer, só submetem tudo de uma vez ao comitar) — se a sua sessão faz o mesmo, declare esta capacidade ausente; se cada escrita já é imediatamente visível para uma leitura seguinte na mesma transação (a maioria dos bancos relacionais/documento), não precisa tocar nela.

Mentir sobre uma capacidade (declarar ligada quando não é) produz uma falha real na primeira requisição, exatamente o que este mecanismo existe para evitar — não force isso para "passar" na checagem de build-time.

Restrições estruturais de consulta (opcional)

Alguns bancos exigem que where: cubra um campo específico (a partition key) para que db.query/db.find/db.exists sejam eficientes — sem isso, a consulta vira uma varredura completa disfarçada. Se o seu provider tiver essa restrição, sobrescreva o default interface member ValidateQueryShape, chamado uma vez por step, no build:

public void ValidateQueryShape(EntityDefinition entity, IReadOnlyCollection<string> whereFields, IReadOnlyCollection<string> sortFields, bool hasSkip)
{
    if (!whereFields.Contains("partitionKeyField"))
    {
        throw new DataProviderConfigurationException(
            $"Entidade '{entity.Name}': 'where:' precisa cobrir a partition key.");
    }
}

O default é um no-op — a maioria dos providers (todo relacional, MongoDB) não tem essa restrição.

Steps próprios: o que db.* não expressa

db.* é o vocabulário portátil, deliberadamente restrito ao que os seis providers do core conseguem expressar em comum — filtro por igualdade, ordenação, paginação. Se o seu banco tem um recurso genuíno que esse vocabulário não alcança (um nível de consistência por consulta, um índice secundário explícito, uma escrita condicional), não force isso dentro de db.* — publique um step próprio, com o mesmo mecanismo [Step("chave")] + registro por DI keyed de qualquer outro step (ver guia de IStep). cassandra.query/cassandra.persist e dynamo.query/dynamo.persist (Pipevine.Data.Cassandra/Pipevine.Data.DynamoDb) são o exemplo real desse padrão: cassandra.query aceita consistency:, algo que só faz sentido para esse banco e que nenhum dos outros cinco tem conceito equivalente. Um step assim não precisa implementar IStepCodeEmitter — cai no caminho interpretado dentro do pipeline compilado, sem prejuízo funcional (extensibilidade nunca é bloqueada por performance).

Outbox transacional (opcional)

Um provider que também quer participar do outbox transacional (usado por publish com mode: outbox) implementa IOutboxStore (escrita, na mesma transação — exige AtomicMultiWrite declarado, já que mode: outbox exige transaction: required). Para também rodar o dispatcher (entregar as mensagens gravadas), implemente IOutboxDispatcherStore (Pipevine.Data.Abstractions.Outbox) no seu IDataSession: EnsureSchemaAsync/ClaimAsync/ MarkDeliveredAsync/RescheduleAsync/DeadLetterAsync. O único contrato — ClaimAsync precisa garantir que duas chamadas concorrentes nunca capturem a mesma linha/documento pendente; como isso é implementado (FOR UPDATE SKIP LOCKED num banco relacional, FindOneAndUpdate atômico no MongoDB) é decisão sua. OutboxDispatcher/OutboxDispatcherHostedService (Pipevine.Data.EntityFrameworkCore) funcionam contra qualquer implementação da interface, não só a relacional.

Para onde ir daqui