Skip to content

Commit 63c754d

Browse files
electricessenceelectricessence
committed
MappedQuery added.
1 parent 49d3c1c commit 63c754d

4 files changed

Lines changed: 586 additions & 2 deletions

File tree

‎Source/Core/MappedQuery.cs‎

Lines changed: 212 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,212 @@
1+
namespace Open.Database.Extensions;
2+
3+
public static partial class CoreExtensions
4+
{
5+
/// <summary>
6+
/// Locks in a reflection-based mapping to model type <typeparamref name="T"/> and returns a deferred
7+
/// <see cref="MappedQuery{T}"/>. Nothing runs until a terminal (an <see cref="IMappedQuery{T}"/>
8+
/// extension such as <c>ToList</c>/<c>FirstOrDefault</c>) is invoked; the mapping is applied per row
9+
/// as the reader streams, so terminals read only the rows they need.
10+
/// </summary>
11+
/// <typeparam name="T">The model type to map the values to (using reflection).</typeparam>
12+
/// <param name="command">The command to read from.</param>
13+
/// <param name="fieldMappingOverrides">
14+
/// An override map of field (property) names to column names; a null column value ignores that field.
15+
/// </param>
16+
/// <returns>A deferred query whose results are mapped to <typeparamref name="T"/>.</returns>
17+
public static MappedQuery<T> Map<T>(
18+
this IExecuteReader command,
19+
params IEnumerable<KeyValuePair<string, string?>>? fieldMappingOverrides)
20+
where T : new()
21+
{
22+
if (command is null) throw new ArgumentNullException(nameof(command));
23+
Contract.EndContractBlock();
24+
25+
return new MappedQuery<T>(new MappedQueryCore<T>(
26+
command,
27+
reader => reader.Results<T>(fieldMappingOverrides),
28+
(reader, token) => Transformer<T>.Create(fieldMappingOverrides).ResultsAsync(reader, token),
29+
0, -1));
30+
}
31+
32+
/// <inheritdoc cref="Map{T}(IExecuteReader, IEnumerable{KeyValuePair{string, string?}}?)"/>
33+
[OverloadResolutionPriority(1)]
34+
public static MappedQuery<T> Map<T>(
35+
this IExecuteReader command)
36+
where T : new()
37+
=> command.Map<T>(null);
38+
39+
/// <inheritdoc cref="Map{T}(IExecuteReader, IEnumerable{KeyValuePair{string, string?}}?)"/>
40+
[OverloadResolutionPriority(-2)]
41+
public static MappedQuery<T> Map<T>(
42+
this IExecuteReader command,
43+
params IEnumerable<(string Field, string? Column)>? fieldMappingOverrides)
44+
where T : new()
45+
=> command.Map<T>(fieldMappingOverrides?.Select(mapping => new KeyValuePair<string, string?>(mapping.Field, mapping.Column)));
46+
47+
/// <summary>
48+
/// Locks in a custom projection and returns a deferred <see cref="MappedQuery{T}"/>. Nothing runs until
49+
/// a terminal is invoked; <paramref name="selector"/> is applied per row as the reader streams.
50+
/// </summary>
51+
/// <typeparam name="T">The type each record is projected to.</typeparam>
52+
/// <param name="command">The command to read from.</param>
53+
/// <param name="selector">The transform applied to each <see cref="IDataRecord"/>.</param>
54+
/// <returns>A deferred query whose results are produced by <paramref name="selector"/>.</returns>
55+
[OverloadResolutionPriority(-1)]
56+
public static MappedQuery<T> Map<T>(
57+
this IExecuteReader command,
58+
Func<IDataRecord, T> selector)
59+
{
60+
if (command is null) throw new ArgumentNullException(nameof(command));
61+
if (selector is null) throw new ArgumentNullException(nameof(selector));
62+
Contract.EndContractBlock();
63+
64+
return new MappedQuery<T>(new MappedQueryCore<T>(
65+
command,
66+
reader => reader.Select(selector),
67+
(reader, token) => reader.SelectAsync(selector, false, token),
68+
0, -1));
69+
}
70+
}
71+
72+
// The shared implementation: holds the command handle (any IExecuteReader), the sync/async projections
73+
// captured at Map-time, and the pending skip/take. The execution itself is the command's own
74+
// ExecuteReader/ExecuteReaderAsync — this type never touches a reader directly. Real async streaming when
75+
// the live reader is a DbDataReader; otherwise the synchronous projection is presented as an async sequence.
76+
readonly struct MappedQueryCore<T>
77+
{
78+
readonly IExecuteReader _command;
79+
readonly Func<IDataReader, IEnumerable<T>> _project;
80+
readonly Func<DbDataReader, CancellationToken, IAsyncEnumerable<T>> _projectAsync;
81+
readonly int _skip;
82+
readonly int _take; // -1 == unbounded.
83+
84+
internal MappedQueryCore(
85+
IExecuteReader command,
86+
Func<IDataReader, IEnumerable<T>> project,
87+
Func<DbDataReader, CancellationToken, IAsyncEnumerable<T>> projectAsync,
88+
int skip, int take)
89+
{ _command = command; _project = project; _projectAsync = projectAsync; _skip = skip; _take = take; }
90+
91+
internal MappedQueryCore<T> WithSkip(int count)
92+
=> count >= 0 ? new(_command, _project, _projectAsync, _skip + count, _take)
93+
: throw new ArgumentOutOfRangeException(nameof(count), count, "Cannot be negative.");
94+
95+
// Take is set exactly once (the staging structs only expose it before any take exists), so this
96+
// just records the limit; Slice applies it after the skip, i.e. Enumerable.Skip(a).Take(b).
97+
internal MappedQueryCore<T> WithTake(int count)
98+
=> count >= 0 ? new(_command, _project, _projectAsync, _skip, count)
99+
: throw new ArgumentOutOfRangeException(nameof(count), count, "Cannot be negative.");
100+
101+
internal void Read(Action<IEnumerable<T>> handler)
102+
{
103+
Func<IDataReader, IEnumerable<T>> project = _project;
104+
int skip = _skip, take = _take;
105+
_command.ExecuteReader(reader => handler(Slice(project(reader), skip, take)), CommandBehavior.SingleResult);
106+
}
107+
108+
internal TResult Read<TResult>(Func<IEnumerable<T>, TResult> handler)
109+
{
110+
Func<IDataReader, IEnumerable<T>> project = _project;
111+
int skip = _skip, take = _take;
112+
return _command.ExecuteReader(reader => handler(Slice(project(reader), skip, take)), CommandBehavior.SingleResult);
113+
}
114+
115+
internal ValueTask ReadAsync(Func<IAsyncEnumerable<T>, CancellationToken, ValueTask> handler, CancellationToken cancellationToken)
116+
{
117+
Func<IDataReader, IEnumerable<T>> project = _project;
118+
Func<DbDataReader, CancellationToken, IAsyncEnumerable<T>> projectAsync = _projectAsync;
119+
int skip = _skip, take = _take;
120+
CancellationToken token = cancellationToken.CanBeCanceled ? cancellationToken : _command.CancellationToken;
121+
return _command.ExecuteReaderAsync(reader => handler(SliceAsync(project, projectAsync, skip, take, reader, token), token), CommandBehavior.SingleResult);
122+
}
123+
124+
internal ValueTask<TResult> ReadAsync<TResult>(Func<IAsyncEnumerable<T>, CancellationToken, ValueTask<TResult>> handler, CancellationToken cancellationToken)
125+
{
126+
Func<IDataReader, IEnumerable<T>> project = _project;
127+
Func<DbDataReader, CancellationToken, IAsyncEnumerable<T>> projectAsync = _projectAsync;
128+
int skip = _skip, take = _take;
129+
CancellationToken token = cancellationToken.CanBeCanceled ? cancellationToken : _command.CancellationToken;
130+
return _command.ExecuteReaderAsync(reader => handler(SliceAsync(project, projectAsync, skip, take, reader, token), token), CommandBehavior.SingleResult);
131+
}
132+
133+
static IEnumerable<T> Slice(IEnumerable<T> sequence, int skip, int take)
134+
{
135+
if (skip > 0) sequence = sequence.Skip(skip);
136+
if (take >= 0) sequence = sequence.Take(take);
137+
return sequence;
138+
}
139+
140+
static IAsyncEnumerable<T> SliceAsync(
141+
Func<IDataReader, IEnumerable<T>> project,
142+
Func<DbDataReader, CancellationToken, IAsyncEnumerable<T>> projectAsync,
143+
int skip, int take, IDataReader reader, CancellationToken token)
144+
{
145+
IAsyncEnumerable<T> sequence = reader is DbDataReader dbReader
146+
? projectAsync(dbReader, token)
147+
: project(reader).ToAsyncEnumerable();
148+
if (skip > 0) sequence = sequence.Skip(skip);
149+
if (take >= 0) sequence = sequence.Take(take);
150+
return sequence;
151+
}
152+
}
153+
154+
/// <summary>A deferred, type-locked query over any <see cref="IExecuteReader"/>. Terminals are extension methods on <see cref="IMappedQuery{T}"/>.</summary>
155+
/// <typeparam name="T">The type each record is mapped to.</typeparam>
156+
public readonly record struct MappedQuery<T> : IMappedQuery<T>
157+
{
158+
readonly MappedQueryCore<T> _core;
159+
internal MappedQuery(MappedQueryCore<T> core) => _core = core;
160+
161+
/// <summary>Skips the first <paramref name="count"/> mapped records.</summary>
162+
public MappedQueryWithSkip<T> Skip(int count) => new(_core.WithSkip(count));
163+
164+
/// <summary>Limits the result to at most <paramref name="count"/> mapped records.</summary>
165+
public MappedQueryWithTake<T> Take(int count) => new(_core.WithTake(count));
166+
167+
/// <inheritdoc />
168+
public void Read(Action<IEnumerable<T>> handler) => _core.Read(handler);
169+
/// <inheritdoc />
170+
public TResult Read<TResult>(Func<IEnumerable<T>, TResult> handler) => _core.Read(handler);
171+
/// <inheritdoc />
172+
public ValueTask ReadAsync(Func<IAsyncEnumerable<T>, CancellationToken, ValueTask> handler, CancellationToken cancellationToken) => _core.ReadAsync(handler, cancellationToken);
173+
/// <inheritdoc />
174+
public ValueTask<TResult> ReadAsync<TResult>(Func<IAsyncEnumerable<T>, CancellationToken, ValueTask<TResult>> handler, CancellationToken cancellationToken) => _core.ReadAsync(handler, cancellationToken);
175+
}
176+
177+
/// <summary>A <see cref="MappedQuery{T}"/> with a pending skip; can still be limited with <see cref="Take(int)"/>.</summary>
178+
/// <typeparam name="T">The type each record is mapped to.</typeparam>
179+
public readonly record struct MappedQueryWithSkip<T> : IMappedQuery<T>
180+
{
181+
readonly MappedQueryCore<T> _core;
182+
internal MappedQueryWithSkip(MappedQueryCore<T> core) => _core = core;
183+
184+
/// <summary>Limits the result to at most <paramref name="count"/> mapped records.</summary>
185+
public MappedQueryWithTake<T> Take(int count) => new(_core.WithTake(count));
186+
187+
/// <inheritdoc />
188+
public void Read(Action<IEnumerable<T>> handler) => _core.Read(handler);
189+
/// <inheritdoc />
190+
public TResult Read<TResult>(Func<IEnumerable<T>, TResult> handler) => _core.Read(handler);
191+
/// <inheritdoc />
192+
public ValueTask ReadAsync(Func<IAsyncEnumerable<T>, CancellationToken, ValueTask> handler, CancellationToken cancellationToken) => _core.ReadAsync(handler, cancellationToken);
193+
/// <inheritdoc />
194+
public ValueTask<TResult> ReadAsync<TResult>(Func<IAsyncEnumerable<T>, CancellationToken, ValueTask<TResult>> handler, CancellationToken cancellationToken) => _core.ReadAsync(handler, cancellationToken);
195+
}
196+
197+
/// <summary>A fully-configured <see cref="MappedQuery{T}"/> with a take (and any preceding skip); terminal-only.</summary>
198+
/// <typeparam name="T">The type each record is mapped to.</typeparam>
199+
public readonly record struct MappedQueryWithTake<T> : IMappedQuery<T>
200+
{
201+
readonly MappedQueryCore<T> _core;
202+
internal MappedQueryWithTake(MappedQueryCore<T> core) => _core = core;
203+
204+
/// <inheritdoc />
205+
public void Read(Action<IEnumerable<T>> handler) => _core.Read(handler);
206+
/// <inheritdoc />
207+
public TResult Read<TResult>(Func<IEnumerable<T>, TResult> handler) => _core.Read(handler);
208+
/// <inheritdoc />
209+
public ValueTask ReadAsync(Func<IAsyncEnumerable<T>, CancellationToken, ValueTask> handler, CancellationToken cancellationToken) => _core.ReadAsync(handler, cancellationToken);
210+
/// <inheritdoc />
211+
public ValueTask<TResult> ReadAsync<TResult>(Func<IAsyncEnumerable<T>, CancellationToken, ValueTask<TResult>> handler, CancellationToken cancellationToken) => _core.ReadAsync(handler, cancellationToken);
212+
}

0 commit comments

Comments
 (0)