adding upsert support

Author: aaronpowell

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - New CI/CD pipeline
 - Primitive support for pagination [#23](https://github.com/aaronpowell/FSharp.CosmosDb/issues/23)
+- Upsert support [#43](https://github.com/aaronpowell/FSharp.CosmosDb/issues/43)
 
 ### Changed
 

README.md

@@ -38,6 +38,22 @@ let insertUsers data =
     |> Cosmos.execAsync
 ```
 
+### Upsert
+
+```fsharp
+open FSharp.CosmosDb
+
+let connStr = "..."
+
+let insertUsers data =
+    connStr
+    |> Cosmos.fromConnectionString
+    |> Cosmos.database "UserDb"
+    |> Cosmos.container "UserContainer"
+    |> Cosmos.upsertMany<User> data
+    |> Cosmos.execAsync
+```
+
 ### Update
 
 ```fsharp
@@ -173,3 +189,4 @@ Add the following settings (globally or in the workspace):
 
 - Zaid Ajaj for the [Npgsql Analyzer](https://github.com/Zaid-Ajaj/Npgsql.FSharp.Analyzer). Without this I wouldn't have been able to work out how to do it (and there's some code lifted from there)
 - [Krzysztof Cieślak](https://twitter.com/k_cieslak) for the amazing Ionide plugin
+- [Isaac Abraham](https://twitter.com/isaac_abraham) for helping fix the parser

src/FSharp.CosmosDb/Cosmos.fs

@@ -68,6 +68,14 @@ module Cosmos =
     let insert<'T> (value: 'T) op =
         Insert { Connection = op; Values = [ value ] }
 
+    // --- INSERT --- //
+
+    let upsertMany<'T> (values: 'T list) op =
+        Upsert { Connection = op; Values = values }
+
+    let upsert<'T> (value: 'T) op =
+        Upsert { Connection = op; Values = [ value ] }
+
     // --- UPDATE --- //
 
     let update<'T> id partitionKey (updater: 'T -> 'T) op =
@@ -116,6 +124,7 @@ module Cosmos =
         | Insert op -> OperationHandling.execInsert getClient op
         | Update op -> OperationHandling.execUpdate getClient op
         | Delete op -> OperationHandling.execDelete getClient op
+        | Upsert op -> OperationHandling.execUpsert getClient op
 
     let execBatchAsync<'T> (op: ContainerOperation<'T>) =
         match op with

src/FSharp.CosmosDb/OperationHandling.fs

@@ -15,9 +15,11 @@ let execQueryInternal (getClient: ConnectionOperation -> CosmosClient) (op: Quer
         let! containerName = connInfo.ContainerName
 
         let db = client.GetDatabase databaseId
+
         let container = db.GetContainer containerName
 
         let! query = op.Query
+
         let qd =
             op.Parameters
             |> List.fold (fun (qd: QueryDefinition) (key, value) -> qd.WithParameter(key, value))
@@ -52,9 +54,59 @@ let execInsert (getClient: ConnectionOperation -> CosmosClient) (op: InsertOp<'T
             let! containerName = connInfo.ContainerName
 
             let db = client.GetDatabase databaseId
+
+            let container = db.GetContainer containerName
+
+            let partitionKey =
+                PartitionKeyAttributeTools.findPartitionKey<'T> ()
+
+            let getPartitionKeyValue single =
+                match partitionKey with
+                | Some propertyInfo ->
+                    let value = propertyInfo.GetValue(single)
+                    Nullable(PartitionKey(value.ToString()))
+                | None -> Nullable()
+
+            return match op.Values with
+                   | [ single ] ->
+                       let pk = getPartitionKeyValue single
+                       [ container.CreateItemAsync<'T>(single, pk)
+                         |> Async.AwaitTask ]
+                   | _ ->
+                       op.Values
+                       |> List.map (fun single ->
+                           let pk = getPartitionKeyValue single
+                           container.CreateItemAsync<'T>(single, pk)
+                           |> Async.AwaitTask)
+        }
+
+    match result with
+    | Some result ->
+        result
+        |> List.map (fun item ->
+            async {
+                let! value = item
+                return value.Value
+            })
+        |> AsyncSeq.ofSeqAsync
+    | None ->
+        failwith "Unable to construct a query as some values are missing across the database, container name and query"
+
+let execUpsert (getClient: ConnectionOperation -> CosmosClient) (op: UpsertOp<'T>) =
+    let connInfo = op.Connection
+    let client = getClient connInfo
+
+    let result =
+        maybe {
+            let! databaseId = connInfo.DatabaseId
+            let! containerName = connInfo.ContainerName
+
+            let db = client.GetDatabase databaseId
+
             let container = db.GetContainer containerName
 
-            let partitionKey = PartitionKeyAttributeTools.findPartitionKey<'T>()
+            let partitionKey =
+                PartitionKeyAttributeTools.findPartitionKey<'T> ()
 
             let getPartitionKeyValue single =
                 match partitionKey with
@@ -66,12 +118,14 @@ let execInsert (getClient: ConnectionOperation -> CosmosClient) (op: InsertOp<'T
             return match op.Values with
                    | [ single ] ->
                        let pk = getPartitionKeyValue single
-                       [ container.CreateItemAsync<'T>(single, pk) |> Async.AwaitTask ]
+                       [ container.UpsertItemAsync<'T>(single, pk)
+                         |> Async.AwaitTask ]
                    | _ ->
                        op.Values
                        |> List.map (fun single ->
                            let pk = getPartitionKeyValue single
-                           container.CreateItemAsync<'T>(single, pk) |> Async.AwaitTask)
+                           container.UpsertItemAsync<'T>(single, pk)
+                           |> Async.AwaitTask)
         }
 
     match result with
@@ -80,7 +134,8 @@ let execInsert (getClient: ConnectionOperation -> CosmosClient) (op: InsertOp<'T
         |> List.map (fun item ->
             async {
                 let! value = item
-                return value.Value })
+                return value.Value
+            })
         |> AsyncSeq.ofSeqAsync
     | None ->
         failwith "Unable to construct a query as some values are missing across the database, container name and query"
@@ -95,19 +150,23 @@ let execUpdate (getClient: ConnectionOperation -> CosmosClient) (op: UpdateOp<'T
             let! containerName = connInfo.ContainerName
 
             let db = client.GetDatabase databaseId
+
             let container = db.GetContainer containerName
 
-            return (container, container.ReadItemAsync(op.Id, PartitionKey op.PartitionKey) |> Async.AwaitTask)
+            return (container,
+                    container.ReadItemAsync(op.Id, PartitionKey op.PartitionKey)
+                    |> Async.AwaitTask)
         }
 
     match result with
-    | Some(container, result) ->
+    | Some (container, result) ->
         [ async {
             let! currentItemResponse = result
 
             let newItem = op.Updater currentItemResponse.Value
 
-            let partitionKey = PartitionKeyAttributeTools.findPartitionKey<'T>()
+            let partitionKey =
+                PartitionKeyAttributeTools.findPartitionKey<'T> ()
 
             let pk =
                 match partitionKey with
@@ -116,7 +175,9 @@ let execUpdate (getClient: ConnectionOperation -> CosmosClient) (op: UpdateOp<'T
                     Nullable(PartitionKey(value.ToString()))
                 | None -> Nullable()
 
-            let! newItemResponse = container.ReplaceItemAsync(newItem, op.Id, pk) |> Async.AwaitTask
+            let! newItemResponse =
+                container.ReplaceItemAsync(newItem, op.Id, pk)
+                |> Async.AwaitTask
 
             return newItemResponse.Value
           } ]
@@ -134,15 +195,19 @@ let execDelete (getClient: ConnectionOperation -> CosmosClient) (op: DeleteOp<'T
             let! containerName = connInfo.ContainerName
 
             let db = client.GetDatabase databaseId
+
             let container = db.GetContainer containerName
 
-            return container.DeleteItemAsync(op.Id, PartitionKey op.PartitionKey) |> Async.AwaitTask
+            return container.DeleteItemAsync(op.Id, PartitionKey op.PartitionKey)
+                   |> Async.AwaitTask
         }
 
     match result with
     | Some result ->
         [ async {
             let! currentItemResponse = result
-            return currentItemResponse.Value } ] |> AsyncSeq.ofSeqAsync
+            return currentItemResponse.Value
+          } ]
+        |> AsyncSeq.ofSeqAsync
 
     | None -> failwith "Unable to read from the container to get the item for updating"

src/FSharp.CosmosDb/Types.fs

@@ -21,6 +21,10 @@ type InsertOp<'T> =
     { Connection: ConnectionOperation
       Values: 'T list }
 
+type UpsertOp<'T> =
+    { Connection: ConnectionOperation
+      Values: 'T list }
+
 type UpdateOp<'T> =
     { Connection: ConnectionOperation
       Id: string
@@ -37,3 +41,4 @@ type ContainerOperation<'T> =
     | Insert of InsertOp<'T>
     | Update of UpdateOp<'T>
     | Delete of DeleteOp<'T>
+    | Upsert of UpsertOp<'T>