Using IAsyncEnumerable with Dapper

Question

We have recently migrated our ASP.NET Core API which uses Dapper to .NET Core 3.1. After the migration, we felt there was an opportunity to use the latest IAsyncEnumerable feature from C# 8 for one of our endpoints.

Here is the pseudocode before the changes:

public async Task<IEnumerable<Item>> GetItems(int id)
{
    var reader = await _connection.QueryMultipleAsync(getItemsSql,
       param: new
       {
           Id = id
       });

    var idFromDb = (await reader.ReadAsync<int?>().ConfigureAwait(false)).SingleOrDefault();
    if (idFromDb == null)
    {
       return null;
    }

    var items = await reader.ReadAsync<Item>(buffered: false).ConfigureAwait(false);

    return Stream(reader, items);
} 

private IEnumerable<Item> Stream(SqlMapper.GridReader reader, IEnumerable<Item> items)
{
    using (reader)
    {
        foreach (var item in items)
        {
            yield return item;
        }
    }     
}

After IAsyncEnumerable code changes:

// Import Nuget pacakage: System.Linq.Async

public async Task<IAsyncEnumerable<Item>> GetItems(int id)
{
    var reader = await _connection.QueryMultipleAsync(getItemsSql,
       param: new
       {
           Id = id
       });

    var idFromDb = (await reader.ReadAsync<int?>().ConfigureAwait(false)).SingleOrDefault();
    if (idFromDb == null)
    {
        return null;
    }

    var items = await reader.ReadAsync<Item>(buffered: false).ConfigureAwait(false);

    return Stream(reader, items);
} 

private IAsyncEnumerable<Item> Stream(SqlMapper.GridReader reader, IEnumerable<Item> items)
{
    using (reader)
    {
       await foreach (var item in items.ToAsyncEnumerable())
       {
           yield return item;
       }
    }
 }

The above approach is to use ToAsyncEnumerable is loosely inspired from this post, but I'm not 100% sure if I'm using it in the right place/ context.

Question:

• The dapper library only returns IEnumerable but can we use ToAsyncEnumerable to convert it into IAsyncEnumerable for async stream like above?

Note: This question looks similar to What happens with returning IEnumerable if used with async/await (streaming data from SQL Server with Dapper)? but I do not think that answers my question.

Answer 1

Update: I wasn't aware of async iterators when I first wrote this answer. Thanks to Theodor Zoulias for pointing it out. In light of that, a much simpler approach is possible:

using var reader = await connection.ExecuteReaderAsync(query, parameters);
var rowParser = reader.GetRowParser<T>();

// Consider using reader.NextResultAsync(). Follow github issue for details:

while (await reader.ReadAsync()) {
    yield return rowParser(reader);
}

Ref: https://github.com/DapperLib/Dapper/issues/1239#issuecomment-1035507322

Original Answer:

Here's an IAsyncEnumerable wrapper I wrote that may help those who want to stream unbuffered data using async/await and also want the power of Dapper's type mapping:

public class ReaderParser<T> : IAsyncEnumerable<T> {
    public ReaderParser(SqlDataReader reader) {
        Reader = reader;
    }
    private SqlDataReader Reader { get; }
    public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default) {
        return new ReaderParserEnumerator<T>(Reader);
    }
}
public class ReaderParserEnumerator<T> : IAsyncEnumerator<T> {
    public ReaderParserEnumerator(SqlDataReader reader) {
        Reader = reader;
        RowParser = reader.GetRowParser<T>();
    }
    public T Current => Reader.FieldCount == 0 ? default(T) : RowParser(Reader);
    private SqlDataReader Reader { get; }
    private Func<IDataReader, T> RowParser { get; }
    public async ValueTask DisposeAsync() {
        await Reader.DisposeAsync();
    }
    public async ValueTask<bool> MoveNextAsync() {
        return await Reader.ReadAsync();
    }
}

Usage:

var reader = await command.ExecuteReaderAsync();
return new ReaderParser<T>(reader);

And then, package System.Linq.Async adds basically all the nice IEnumerable extensions you know and love, e.g. in my usage:

var streamData = await repo.GetDataStream();
var buffer = await streamData.Take(BATCH_SIZE).ToListAsync();

Answer 2

This feature is directly available in Dapper since its 2.0.138 release. Use its QueryUnbufferedAsync DbConnection extension or the GridReader.ReadUnbufferedAsync method on batches.

For previous versions of Dapper:

Dave's answer gives all that is needed, but is not straightforward enough for me.

So, elaborated from it, here is an extension method for obtaining an IAsyncEnumerable with Dapper:

/// <summary>
/// Asynchronously enumerates the results of a query.
/// </summary>
/// <typeparam name="T">The type of result to return.</typeparam>
/// <param name="cnn">The connection to query on.</param>
/// <param name="sql">The SQL to execute for the query.</param>
/// <param name="param">The parameters to pass, if any.</param>
/// <param name="transaction">The transaction to use, if any.</param>
/// <returns>An asynchronous enumerator of the results.</returns>
/// <remarks>See <see href="https://stackoverflow.com/a/66723553/1178314"/> and
/// <see href="https://github.com/DapperLib/Dapper/issues/1239#issuecomment-1035507322"/>.</remarks>
public static async IAsyncEnumerable<T> EnumerateAsync<T>(this DbConnection cnn, string sql, object param = null, IDbTransaction transaction = null)
{
    await using var reader = await cnn.ExecuteReaderAsync(sql, param, transaction).ConfigureAwait(false);
    var rowParser = reader.GetRowParser<T>();
    while (await reader.ReadAsync().ConfigureAwait(false))
    {
        yield return rowParser(reader);
    }
    while (await reader.NextResultAsync().ConfigureAwait(false)) { }
}

Note that you must use a DbConnection, not an IDbConnection. Otherwise, reader.ReadAsync will not be available.

Then in your code you can do:

var asyncEnumerable = connection.EnumerateAsync<YourDto>("select ...", yourParameters);