src/cosmos: a cosmos sql source with a query file, day window and shaping (0.8.0)

DataAgent/CAPABILITIES.md

@@ -26,6 +26,7 @@ owned options are the keys the adapter interprets itself. other arguments belong
 |---|---|---|
 | `src/sql` | none | `Invoke-Sqlcmd` |
 | `src/csv` | none | `Import-Csv` |
+| `src/cosmos` | `Endpoint`, `Key`, `Database`, `Container`, `Query` or `InputFile`, `Parameters`, `Days`, `Fresh`, `Formats`, `Map`, `Sort`, `Drop` | `Invoke-WebRequest` |
 | `fmt/csv` | `Path`, `Encoding`, `StripQuotes` | `ConvertTo-Csv` |
 | `fmt/xlsx` | `Path` | `Export-Excel` |
 | `fmt/custom` | `Module`, `Path` | `ConvertTo-Custom` |
@@ -36,6 +37,7 @@ owned options are the keys the adapter interprets itself. other arguments belong
 
 - **sql:** arguments pass to Invoke-Sqlcmd, including connection strings and authentication settings. DataTable results emit their rows; empty tables emit nothing. multiple result sets are flattened, not separate artifacts. use compatible schemas.
 - **csv source:** arguments pass directly to Import-Csv. relative paths resolve in the job directory.
+- **cosmos source:** a cosmos sql query over rest with the account key (Key, usually `env:`), no sdk needed. it needs PowerShell 7.5 (the rest of DataAgent runs on 7.4). Endpoint is the account url, Database and Container name the collection. the query is Query, or InputFile read from the job directory. Parameters is a hashtable of `@name` to value, and the source adds `@day` (AS_OF, or yesterday), with `@from` and `@to` the Days days (1 by default) ending with it, half open, when the query uses them. Fresh `{ Query, MaxAgeHours }` stops the run when the query's one value, a utc time, is older than that. every page is read and each record keeps its values as stored (an iso date stays a string). cosmos leaves a missing value out of a record, so every record gets every column any record has, empty when missing; write `?? ""` in the query to keep a column that no record has. then Formats `{ column: .net date format }` writes a column of stored dates, Map `{ column: { code: name } }` swaps a trimmed code for a name and keeps the code when it has none, Sort orders by columns, and Drop removes columns that were only there to sort on. the rest gateway serves no aggregates and no cross-partition order by, so sort here, not in the query.
 - **csv format:** Path is the output; Encoding defaults to utf8NoBOM. all remaining arguments pass to ConvertTo-Csv, including Delimiter, UseQuotes and NoHeader. StripQuotes removes *every* double quote, including data. unquoted modes can lose meaning with commas/newlines. source column order is retained.
 - **xlsx:** all arguments pass to Export-Excel, including Path and layout options. an existing file at Path is replaced, never written into, so a shorter run leaves no rows from the last one. verify cells/layout, not ZIP hashes. AutoSize may need native support on non-Windows hosts.
 - **custom format bridge:** Module is a path to a module exporting `ConvertTo-Custom($dt)`. the bridge passes a DataTable and saves returned text to Path as UTF-8 without BOM. this bridge is for existing text converters, not binary ZIP output.

DataAgent/DataAgent.psd1

@@ -1,6 +1,6 @@
 @{
     RootModule = 'DataAgent.psm1'
-    ModuleVersion = '0.7.5'
+    ModuleVersion = '0.8.0'
     GUID = '32662c0b-0a5d-49ac-8c62-8f7ae05f79c0'
     Author = 'Roy Ashbrook'
     Copyright = '(c) 2026 Roy Ashbrook. MIT License.'
@@ -10,13 +10,13 @@
     CmdletsToExport = @()
     VariablesToExport = @()
     AliasesToExport = @()
-    FileList = @('DataAgent.psd1', 'DataAgent.psm1', 'LICENSE', 'CAPABILITIES.md', 'src/sql.ps1', 'src/csv.ps1', 'fmt/csv.ps1', 'fmt/xlsx.ps1', 'fmt/custom.ps1', 'dst/email.ps1', 'dst/sftp.ps1', 'dst/ftp.ps1', 'dst/ftps.ps1')
+    FileList = @('DataAgent.psd1', 'DataAgent.psm1', 'LICENSE', 'CAPABILITIES.md', 'src/sql.ps1', 'src/csv.ps1', 'src/cosmos.ps1', 'fmt/csv.ps1', 'fmt/xlsx.ps1', 'fmt/custom.ps1', 'dst/email.ps1', 'dst/sftp.ps1', 'dst/ftp.ps1', 'dst/ftps.ps1')
     PrivateData = @{
         PSData = @{
             Tags = @('ETL', 'Data', 'CSV', 'Automation', 'PSEdition_Core', 'Windows', 'Linux', 'MacOS')
             LicenseUri = 'https://github.com/royashbrook/DataAgent/blob/main/LICENSE'
             ProjectUri = 'https://github.com/royashbrook/DataAgent'
-            ReleaseNotes = 'Forward optional filename_date_pattern to Clear-Files 1.2.0.0 for caller-selected dates anywhere in filenames. Missing keys preserve mtime/prefix behavior. Cleanup placement, adapters and cache policy are unchanged.'
+            ReleaseNotes = 'src/cosmos: a cosmos sql query over rest with the account key, from Query or an InputFile, with @day, @from and @to for the day a run reports, an optional freshness check, and the shaping a cosmos query cannot do (date formats, code maps, sort, dropped sort columns). No other change.'
         }
     }
 }

DataAgent/src/cosmos.ps1

@@ -0,0 +1,66 @@
+#requires -Version 7.5
+# 7.5 for ConvertFrom-Json -DateKind; the rest of DataAgent runs on 7.4
+param($Data, [hashtable] $Options)
+# a cosmos sql query over rest with the account key, one post per page: no sdk on the runner
+$link = "dbs/$($Options.Database)/colls/$($Options.Container)"
+function Read-Cosmos([string] $Query, [hashtable] $Parameters) {
+    $list = @(if ($Parameters) { foreach ($p in $Parameters.GetEnumerator()) { @{ name = $p.Key; value = $p.Value } } })
+    $body = @{ query = $Query; parameters = $list } | ConvertTo-Json -Depth 10 -Compress
+    $continuation = $null
+    do {
+        $date = [DateTime]::UtcNow.ToString('R').ToLower()
+        $hmac = [Security.Cryptography.HMACSHA256]::new([Convert]::FromBase64String($Options.Key))
+        try { $signature = [Convert]::ToBase64String($hmac.ComputeHash([Text.Encoding]::UTF8.GetBytes("post`ndocs`n$link`n$date`n`n"))) } finally { $hmac.Dispose() }
+        $headers = @{
+            Authorization = [Uri]::EscapeDataString("type=master&ver=1.0&sig=$signature"); 'x-ms-date' = $date; 'x-ms-version' = '2018-12-31'
+            'x-ms-documentdb-isquery' = 'true'; 'x-ms-documentdb-query-enablecrosspartition' = 'true'
+        }
+        if ($continuation) { $headers['x-ms-continuation'] = $continuation }
+        $response = Invoke-WebRequest -Method Post -Uri "$($Options.Endpoint.TrimEnd('/'))/$link/docs" -Headers $headers -Body $body -ContentType 'application/query+json'
+        # a source hands back what is stored: an iso date stays its string, not a local DateTime
+        ($response.Content | ConvertFrom-Json -DateKind String).Documents
+        $continuation = "$($response.Headers['x-ms-continuation'])"
+    } while ($continuation)
+}
+
+# Fresh: a query whose one value is when the container was last loaded, and the most hours that may be
+if ($Options.Fresh) {
+    $last = @(Read-Cosmos $Options.Fresh.Query)[0]
+    if (-not $last) { throw "cosmos freshness: '$($Options.Fresh.Query)' returned nothing." }
+    $age = [datetime]::UtcNow - [datetime]::Parse($last, [cultureinfo]::InvariantCulture, 'AdjustToUniversal,AssumeUniversal')
+    "cosmos last loaded $last, $([math]::Round($age.TotalMinutes)) minutes ago" | Write-Host
+    if ($age.TotalHours -gt $Options.Fresh.MaxAgeHours) { throw "cosmos last loaded $([math]::Round($age.TotalHours, 1)) hours ago, over $($Options.Fresh.MaxAgeHours)." }
+}
+
+# @day is the day the run reports (AS_OF, or yesterday), @from and @to the Days days ending with it, half open
+$query = if ($Options.InputFile) { Get-Content -LiteralPath $Options.InputFile -Raw } else { $Options.Query }
+$day = if ($env:AS_OF) { [datetime]::ParseExact($env:AS_OF, 'yyyy-MM-dd', [cultureinfo]::InvariantCulture) } else { (Get-Date).Date.AddDays(-1) }
+$days = if ($Options.Days) { [int]$Options.Days } else { 1 }
+$parameters = @{ '@day' = $day.ToString('yyyy-MM-dd'); '@from' = $day.AddDays(1 - $days).ToString('yyyy-MM-dd'); '@to' = $day.AddDays(1).ToString('yyyy-MM-dd') }
+$used = @{}; foreach ($k in $parameters.Keys) { if ($query -match "$k\b") { $used[$k] = $parameters[$k] } }
+if ($Options.Parameters) { foreach ($p in $Options.Parameters.GetEnumerator()) { $used[$p.Key] = $p.Value } }
+$documents = @(Read-Cosmos $query $used)
+"cosmos for $($parameters['@day']): $($documents.Count) records" | Write-Host
+if (-not $documents.Count) { return }
+
+# a value cosmos leaves out comes back as no property at all, so every record gets every column, empty when missing
+$columns = [Collections.Generic.List[string]]::new()
+foreach ($d in $documents) { foreach ($p in $d.PSObject.Properties) { if (-not $columns.Contains($p.Name)) { $columns.Add($p.Name) } } }
+$en = [cultureinfo]::InvariantCulture
+$rows = foreach ($d in $documents) {
+    $row = [ordered]@{}
+    foreach ($c in $columns) {
+        $v = $d.$c
+        if ($null -eq $v) { $v = '' }
+        # Formats: a column of stored iso dates written in a .net date format
+        elseif ($Options.Formats -and $Options.Formats.Contains($c) -and "$v") { $v = [datetime]::Parse($v, $en).ToString($Options.Formats[$c], $en) }
+        # Map: a column's codes to the names a customer knows, matched trimmed, the stored value when the code has none
+        elseif ($Options.Map -and $Options.Map.Contains($c) -and $Options.Map[$c]["$v".Trim()]) { $v = $Options.Map[$c]["$v".Trim()] }
+        $row[$c] = $v
+    }
+    [pscustomobject]$row
+}
+# Sort: columns to order by, in order; Drop: columns that were only there to sort on
+if ($Options.Sort) { $rows = $rows | Sort-Object -Property @($Options.Sort) -Stable }
+if ($Options.Drop) { $rows = $rows | Select-Object -Property * -ExcludeProperty @($Options.Drop) }
+$rows

README.md

@@ -3,21 +3,21 @@
 the repeated part of a feed job: get data, format it, send it. use the tools you already use.
 
 ```powershell
-Install-Module DataAgent -RequiredVersion 0.7.5 -Scope CurrentUser
+Install-Module DataAgent -RequiredVersion 0.8.0 -Scope CurrentUser
 ```
 
 ## the job
 
 ```powershell
-Import-Module DataAgent -RequiredVersion 0.7.5
+Import-Module DataAgent -RequiredVersion 0.8.0
 Invoke-DataAgent "$PSScriptRoot/settings.json"
 ```
 
 that is the whole job's `.ps1`. the run works in the settings file's folder. a secret stays out of the file: write `"env:NAME"` as the value and it is read from the environment when the run starts, and `{ "username": ..., "password": ... }` becomes a PSCredential. a script can still pass a hashtable instead of a path. [settings.json](examples/settings.json) and [job.ps1](examples/job.ps1) show SQL + CSV + email.
 
 ## what it supports
 
-read the [capability brief](DataAgent/CAPABILITIES.md) for the config contract, all nine built-in adapters, owned options versus helper pass-throughs, custom scripts, dependencies and limits. the same file ships in the gallery module, so an agent or person can inspect the installed contract without reading code or running a job.
+read the [capability brief](DataAgent/CAPABILITIES.md) for the config contract, all ten built-in adapters, owned options versus helper pass-throughs, custom scripts, dependencies and limits. the same file ships in the gallery module, so an agent or person can inspect the installed contract without reading code or running a job.
 
 the runner sets location to the settings file's folder, the calling script's directory, or the config's `directory` (for a module that wraps the runner), logs to screen and `yyyyMMdd.log`, cleans up when configured, then gets, formats and sends. empty results send nothing. `-WhatIf` skips the whole run. custom adapters need no core edit.
 

examples/job.ps1

@@ -1,2 +1,2 @@
-Import-Module DataAgent -RequiredVersion 0.7.5 -ErrorAction Stop
+Import-Module DataAgent -RequiredVersion 0.8.0 -ErrorAction Stop
 Invoke-DataAgent "$PSScriptRoot/settings.json"

testing/DataAgent.Test/DataAgent.Test.psd1

@@ -1,6 +1,6 @@
 @{
     RootModule = 'DataAgent.Test.psm1'
-    ModuleVersion = '0.7.5'
+    ModuleVersion = '0.8.0'
     GUID = '3d16fcb0-a991-4c65-a6e4-3bfcd360688d'
     Author = 'Roy Ashbrook'
     Copyright = '(c) 2026 Roy Ashbrook. MIT License.'
@@ -10,7 +10,7 @@
     CmdletsToExport = @()
     VariablesToExport = @()
     AliasesToExport = @()
-    RequiredModules = @(@{ ModuleName = 'DataAgent'; RequiredVersion = '0.7.5' })
+    RequiredModules = @(@{ ModuleName = 'DataAgent'; RequiredVersion = '0.8.0' })
     FileList = @('DataAgent.Test.psd1', 'DataAgent.Test.psm1', 'LICENSE', 'synthetic.csv')
     PrivateData = @{ PSData = @{ Tags = @('ETL', 'DataAgent'); ProjectUri = 'https://github.com/royashbrook/DataAgent'; LicenseUri = 'https://github.com/royashbrook/DataAgent/blob/main/LICENSE' } }
 }

tests/acceptance.ps1

@@ -370,6 +370,7 @@ Assert ((Get-Content "$root/list/login.txt") -eq 'PSCredential|synthetic user|li
 Remove-Item Env:DATAAGENT_TEST_SOURCE, Env:DATAAGENT_TEST_USER
 
 . "$PSScriptRoot/transfers.ps1"
+. "$PSScriptRoot/cosmos.ps1"
 
 foreach ($package in @('DataAgent','testing/DataAgent.Test')) {
     $name = Split-Path $package -Leaf
@@ -380,7 +381,7 @@ foreach ($package in @('DataAgent','testing/DataAgent.Test')) {
 }
 Remove-Module DataAgent -Force
 $env:PSModulePath = (@("$root/staged", $priorModules) -join [IO.Path]::PathSeparator)
-Import-Module DataAgent.Test -RequiredVersion 0.7.5
+Import-Module DataAgent.Test -RequiredVersion 0.8.0
 $result = @(Test-DataAgent)
 Assert (@($result | Where-Object { $_ -is [IO.FileInfo] -and $_.Name -eq 'output.csv' }).Count -eq 1) 'staged optional test package exercises bundled formatter'
 $env:PSModulePath = $priorModules

tests/cosmos.ps1

@@ -0,0 +1,75 @@
+# The cosmos source over a stubbed Invoke-WebRequest: two pages, the auth header, the query body, a refusal.
+$global:cosmosCalls = [Collections.Generic.List[object]]::new()
+$global:cosmosKey = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes('synthetic-key-for-the-test-only!'))
+function global:Invoke-WebRequest {
+    param($Method, $Uri, $Headers, $Body, $ContentType)
+    $global:cosmosCalls.Add(@{ Method = $Method; Uri = $Uri; Headers = $Headers; Body = $Body; ContentType = $ContentType })
+    if ($Headers['x-ms-date'] -eq 'refuse') { throw 'Unauthorized' }
+    if ($Uri -match 'refusing') { throw 'Unauthorized' }
+    $page = if ($Headers.ContainsKey('x-ms-continuation')) { 2 } else { 1 }
+    $documents = @(@{ id = "$page-a"; billto = 'SYN'; gross = $page * 10; at = '2026-09-30T04:13:00Z' }, @{ id = "$page-b"; billto = 'SYN'; gross = $page * 10 + 1 })
+    $out = [pscustomobject]@{ Content = (@{ Documents = $documents; _count = 2 } | ConvertTo-Json -Depth 5 -Compress); Headers = @{} }
+    if ($page -eq 1) { $out.Headers['x-ms-continuation'] = @('token-page-2') }
+    $out
+}
+try {
+    $cfg = Config 'cosmos'
+    $cfg.src = @{ adapter = 'cosmos'; args = @{ Endpoint = 'https://synthetic.documents.invalid/'; Key = $global:cosmosKey; Database = 'bc'; Container = 'orders'
+        Query = 'select c.id, c.billto, c.gross from c where c.billto = @billto'; Parameters = @{ '@billto' = 'SYN' } } }
+    $null = Run 'cosmos' $cfg
+    $rows = @(Import-Csv "$root/cosmos/output.csv")
+    Assert ($rows.Count -eq 4 -and $rows[0].id -eq '1-a' -and $rows[3].gross -eq '21') 'cosmos: both pages become rows, in order'
+    Assert ($rows[0].at -ceq '2026-09-30T04:13:00Z') 'cosmos: an iso date comes through as the stored string'
+    Assert ($global:cosmosCalls.Count -eq 2 -and $global:cosmosCalls[1].Headers['x-ms-continuation'] -eq 'token-page-2') 'cosmos: the second post carries the continuation token'
+    $first = $global:cosmosCalls[0]
+    Assert ($first.Uri -eq 'https://synthetic.documents.invalid/dbs/bc/colls/orders/docs' -and $first.ContentType -eq 'application/query+json') 'cosmos: query posts to the container docs link'
+    $sent = $first.Body | ConvertFrom-Json
+    Assert ($sent.query -like 'select c.id*' -and $sent.parameters[0].name -eq '@billto' -and $sent.parameters[0].value -eq 'SYN') 'cosmos: query and parameters in the body'
+    $date = $first.Headers['x-ms-date']
+    $hmac = [Security.Cryptography.HMACSHA256]::new([Convert]::FromBase64String($global:cosmosKey))
+    $expected = [Uri]::EscapeDataString("type=master&ver=1.0&sig=" + [Convert]::ToBase64String($hmac.ComputeHash([Text.Encoding]::UTF8.GetBytes("post`ndocs`ndbs/bc/colls/orders`n$date`n`n"))))
+    $hmac.Dispose()
+    Assert ($first.Headers['Authorization'] -eq $expected -and $date -ceq $date.ToLower()) 'cosmos: master key signature over verb, resource, link and the lowercase date'
+    $cfg.src.args.Endpoint = 'https://refusing.documents.invalid'
+    Refuses { Run 'cosmos' $cfg } 'Unauthorized'
+} finally { Remove-Item function:global:Invoke-WebRequest, variable:global:cosmosCalls }
+
+# The query file, the day window, freshness, and the shaping a cosmos query cannot do itself, over a stub that
+# answers the freshness query with a marker and anything else with three records in no useful order.
+$global:cosmosBodies = [Collections.Generic.List[object]]::new()
+$global:cosmosMarker = [datetime]::UtcNow.AddHours(-1).ToString('o')
+function global:Invoke-WebRequest {
+    param($Method, $Uri, $Headers, $Body, $ContentType)
+    $sent = $Body | ConvertFrom-Json
+    $global:cosmosBodies.Add($sent)
+    $documents = if ($sent.query -match '_sync') { @(@($global:cosmosMarker) | Where-Object { $_ }) } else {
+        @(
+            [ordered]@{ _order = 2; _line = 1; Name = 'b'; When = '2026-10-13T03:25:00'; Code = ' EXX ' }
+            [ordered]@{ _order = 1; _line = 2; Name = 'a'; When = ''; Code = 'ZZZ' }
+            [ordered]@{ _order = 1; _line = 1; Name = 'a2'; Code = 'EXX'; Extra = 'x' }
+        )
+    }
+    [pscustomobject]@{ Content = (@{ Documents = $documents } | ConvertTo-Json -Depth 5 -Compress); Headers = @{} }
+}
+try {
+    $env:AS_OF = '2026-10-13'
+    $cfg = Config 'cosmos-shape'
+    'select value c from c where c.billto = @billto and c.day >= @from and startswith(c.done, @day)' | Set-Content "$root/cosmos-shape/get-data.sql"
+    $fresh = @{ Query = "select value c.lastSuccess from c where c.id = '_sync'"; MaxAgeHours = 24 }
+    $cfg.src = @{ adapter = 'cosmos'; args = @{ Endpoint = 'https://synthetic.documents.invalid'; Key = $global:cosmosKey; Database = 'bc'; Container = 'orders'
+        InputFile = 'get-data.sql'; Parameters = @{ '@billto' = 'SYN' }; Days = 3; Fresh = $fresh
+        Formats = @{ When = 'MM/dd/yyyy HH:mm' }; Map = @{ Code = @{ EXX = 'Exxon' } }; Sort = @('_order', '_line'); Drop = @('_order', '_line') } }
+    $null = Run 'cosmos-shape' $cfg
+    $query = $global:cosmosBodies | Where-Object { $_.query -notmatch '_sync' } | Select-Object -First 1
+    $values = @{}; foreach ($p in $query.parameters) { $values[$p.name] = $p.value }
+    Assert ($query.query -like 'select value c from c where c.billto = @billto*') 'cosmos: InputFile is the query'
+    Assert ($values['@day'] -eq '2026-10-13' -and $values['@from'] -eq '2026-10-11' -and $values['@billto'] -eq 'SYN' -and -not $values.ContainsKey('@to')) 'cosmos: @day is AS_OF, @from opens a Days window, and only the parameters the query uses are sent'
+    $lines = @(Get-Content "$root/cosmos-shape/output.csv")
+    Assert ($lines[0] -ceq '"Name","When","Code","Extra"') 'cosmos: every column any record has, in first-seen order, less the dropped sort columns'
+    Assert ($lines[1] -ceq '"a2","","Exxon","x"' -and $lines[2] -ceq '"a","","ZZZ",""' -and $lines[3] -ceq '"b","10/13/2026 03:25","Exxon",""') 'cosmos: sorted, a missing value empty, dates formatted, codes mapped trimmed, an unmapped code kept'
+    $global:cosmosMarker = [datetime]::UtcNow.AddHours(-30).ToString('o')
+    Refuses { Run 'cosmos-shape' $cfg } 'over 24'
+    $global:cosmosMarker = $null
+    Refuses { Run 'cosmos-shape' $cfg } 'returned nothing'
+} finally { Remove-Item Env:AS_OF -ErrorAction Ignore; Remove-Item function:global:Invoke-WebRequest, variable:global:cosmosBodies, variable:global:cosmosMarker }
+Remove-Item variable:global:cosmosKey