如何在 CosmosDB SDK v3.0 异步查询中使用 LINQ？

Question

我一直在关注官方文档： https://docs.microsoft.com/en-us/azure/cosmos-db/sql-api-get-started#Query

但我不知道如何在 SQL 字符串上正确使用 LINQ 表达式。我尝试了 GetItemLinqQueryable，但我不知道是否是正确的使用方法。也不是异步的。

var db = Client.GetDatabase(databaseId);
var container = db.GetContainer(containerId);

var q = container.GetItemLinqQueryable<Person>();
var result = q.Where(p => p.Name == "Name").ToList();

这是在 Cosmos v3 中使用 LINQ 的正确方法，以及如何使其异步？

Answer 1

您将使用ToFeedIterator() 和FeedIterator<T>.ReadNextAsync()。

var db = Client.GetDatabase(databaseId);
var container = db.GetContainer(containerId);

var q = container.GetItemLinqQueryable<Person>();
var iterator = q.Where(p => p.Name == "Name").ToFeedIterator();
var results = await iterator.ReadNextAsync();

Answer 2

如果您的应用程序遵循分层架构，并且您希望让您的域层完全控制查询，则可以使用实现 IAsyncEnumerable 的自定义 IQueryProvider 包装 cosmos IQueryable<Person>，例如

通过这样做，您可以隐藏从域层异步迭代结果的实现细节。

持久层

public class PersonRepository
{
    public IQueryable<Person> Persons => _cosmosContainer.GetItemLinqQueryable<Person>().ToCosmosAsyncQueryable();
}

域层

var persons = await _personRepository.Persons
    .Where(p => p.Name == "Name")
    .AsAsyncQueryable()
    .ToListAsync(cancellationToken);

• ToListAsync 可从 System.Linq.Async 获得，可以从您的域层引用

域层扩展

public static IAsyncEnumerable<T> AsAsyncQueryable<T>(this IQueryable<T> queryable)
{
    return (IAsyncEnumerable<T>)queryable;
}

持久层扩展

internal static class CosmosAsyncQueryableExtensions
{
    internal static IQueryable<T> ToCosmosAsyncQueryable<T>(this IOrderedQueryable<T> source)
    {
        return new CosmosAsyncQueryable<T>(source);
    }
}

internal class CosmosAsyncQueryable<TResult> : IEnumerable<TResult>, IQueryable<TResult>, IAsyncEnumerable<TResult>
{
    private readonly IQueryable<TResult> _queryable;

    public CosmosAsyncQueryable(IQueryable<TResult> queryable)
    {
        _queryable = queryable;
        Provider = new CosmosAsyncQueryableProvider(queryable.Provider);
    }

    public Type ElementType => typeof(TResult);

    public Expression Expression => _queryable.Expression;

    public IQueryProvider Provider { get; }

    public IEnumerator<TResult> GetEnumerator() => _queryable.GetEnumerator();

    IEnumerator IEnumerable.GetEnumerator() => _queryable.GetEnumerator();

    public async IAsyncEnumerator<TResult> GetAsyncEnumerator(CancellationToken cancellationToken = default)
    {
        var iterator = _queryable.ToFeedIterator();

        while (iterator.HasMoreResults)
        {
            foreach (var item in await iterator.ReadNextAsync(cancellationToken))
            {
                yield return item;
            }
        }
    }
}

internal class CosmosAsyncQueryableProvider : IQueryProvider
{
    private readonly IQueryProvider _provider;

    public CosmosAsyncQueryableProvider(IQueryProvider provider) => _provider = provider;

    public IQueryable<TElement> CreateQuery<TElement>(Expression expression) =>
        new CosmosAsyncQueryable<TElement>(_provider.CreateQuery<TElement>(expression));

    public IQueryable CreateQuery(Expression expression) => CreateQuery<object>(expression);

    public object Execute(Expression expression) => _provider.Execute(expression);

    public TResult Execute<TResult>(Expression expression) => _provider.Execute<TResult>(expression);
}