.net-core - 如何在 FSharp.Data.GraphQL 中实现批处理？

Question

我正在使用 F# 和 .NET Core 构建一个 GraphQL 服务器。为了实现批处理和地址 N+1 选择，我正在构建一个数据加载器。 Facebook 的dataloader使用 Node.js 事件循环滴答 来收集和分派(dispatch)批量请求。

但是，这种机制在 .NET Core 中不可用。我知道我可以从可以手动调用的数据加载器实例中实现 run/dispatch 方法。但是，在独立执行的解析器中很难做到这一点。所以我需要一些自动调度机制来运行批处理请求。

关于如何实现它有什么建议吗？

Answer 1

这是一个很好的问题，我最近问自己。

有一些用于 F# 和 .NET 的“数据加载器”类型库，但是如果您还使用 FSharp.Data.GraphQL，那么能够很好集成的解决方案就更少了。

请注意，“Haxl”方法不能(轻松地)用于 FSharp.Data.GraphQL。这是因为 Haxl 类型必须集成到 GraphQL 查询模型中，但 FSharp.Data.GraphQL 只理解同步和 异步。

我能找到的最合适的实现在 FSharp.Core.Extensions 中.这是一个相当新的库，但它质量很高并且已获得 Apache 2.0 许可。

我确信有很多方法可以将它集成到 FSharp.Data.GraphQL 中，但是我的首选方法是将数据加载器放入模式的根值中。这允许树下的所有 GraphQL 解析器访问它。

我认为最好的解释方式是举个例子。

这里我们有一个“People”域，可以有零个或多个“followers”，他们也是“People”。每个人都有一个全局唯一的 ID。人与人之间的关注者有很大的重叠，所以一个天真的解决方案可能会重复地重新获取相同的数据。我们的数据库层可以在一次查询中获取许多人的记录，因此我们希望尽可能利用这一点。

您可以将此代码粘贴到 .fsx 文件中并运行它。依赖项由 Paket 获取。

paket.dependencies

generate_load_scripts: true

source https://www.nuget.org/api/v2
source https://api.nuget.org/v3/index.json

storage: none
framework: net5.0, netstandard2.1

nuget FSharp.Core 5.0.0
nuget FSharp.Data.GraphQL.Server 1.0.7

github Horusiath/fsharp.core.extensions:0ff5753bb6f232e0ef3c446ddcc72345b74174ca

DataLoader.fsx

#load ".paket/load/net50/FSharp.Data.GraphQL.Server.fsx"

#load "paket-files/Horusiath/fsharp.core.extensions/src/FSharp.Core.Extensions/Prolog.fs"
#load "paket-files/Horusiath/fsharp.core.extensions/src/FSharp.Core.Extensions/AsyncExtensions.fs"

type Person =
  {
    ID : string
    Name : string
  }

// Mocks a real database access layer
module DB =

  // Used to avoid interleaving of printfn calls during async execution
  let private logger = MailboxProcessor.Start (fun inbox -> async {
    while true do
      let! message = inbox.Receive()

      printfn "DB: %s" message
  })

  let private log x =
    logger.Post(x)

  // Our data-set
  let private people =
    [
      { ID = "alice"; Name = "Alice" }, [ "bob"; "charlie"; "david"; "fred" ]
      { ID = "bob"; Name = "Bob" }, [ "charlie"; "david"; "emily" ]
      { ID = "charlie"; Name = "Charlie" }, [ "david" ]
      { ID = "david"; Name = "David" }, [ "emily"; "fred" ]
      { ID = "emily"; Name = "Emily" }, [ "fred" ]
      { ID = "fred"; Name = "Fred" }, []
    ]
    |> Seq.map (fun (p, fs) -> p.ID, (p, fs))
    |> Map.ofSeq

  let fetchPerson id =
    async {
      log $"fetchPerson {id}"

      match people |> Map.find id with
      | (x, _) -> return x
    }

  let fetchPersonBatch ids =
    async {
      let idsString = String.concat "; " ids
      log $"fetchPersonBatch [ {idsString} ]"

      return
        people
        |> Map.filter (fun k _ -> Set.contains k ids)
        |> Map.toSeq
        |> Seq.map (snd >> fst)
        |> Seq.toList
    }

  let fetchFollowers id =
    async {
      log $"fetchFollowers {id}"

      match people |> Map.tryFind id with
      | Some (_, followerIDs) -> return followerIDs
      | _ -> return []
    }





// GraphQL type definitions

open FSharp.Core
open FSharp.Data.GraphQL
open FSharp.Data.GraphQL.Types

#nowarn "40"

[<NoComparison>]
type Root =
  {
    FetchPerson : string -> Async<Person>
    FetchFollowers : string -> Async<string list>
  }

let rec personType =
  Define.Object(
    "Person",
    fun () -> [
      Define.Field("id", ID, fun ctx p -> p.ID)
      Define.Field("name", String, fun ctx p -> p.Name)
      Define.AsyncField("followers", ListOf personType, fun ctx p -> async {
        let root = ctx.Context.RootValue :?> Root

        let! followerIDs = root.FetchFollowers p.ID

        let! followers =
          followerIDs
          |> List.map root.FetchPerson
          |> Async.Parallel

        return Seq.toList followers
      })
    ])

let queryRoot = Define.Object("Query", [
  Define.AsyncField(
    "person",
    personType,
    "Fetches a person by ID",
    [
      Define.Input("id", ID)
    ],
    fun ctx root -> async {
      let id = ctx.Arg("id")

      return! root.FetchPerson id
    })
])

// Construct the schema once to cache it
let schema = Schema(queryRoot)




// Run an example query...
// Here we fetch the followers of the followers of the followers of `alice`
// This query offers many optimization opportunities to the data-loader

let query = """
  query Example {
    person(id: "alice") {
      id
      name
      followers {
        id
        name
        followers {
          id
          name
          followers {
            id
            name
          }
        }
      }
    }
  }
  """

let executor = Executor(schema)

async {
  // Construct a data-loader for fetch person requests
  let fetchPersonBatchFn (requests : Set<string>) =
    async {
      let! people =
        requests
        |> DB.fetchPersonBatch

      let responses =
        Seq.zip requests people
        |> Map.ofSeq

      return responses
    }

  let fetchPersonContext = DataLoader.context ()
  let fetchPersonLoader = DataLoader.create fetchPersonContext fetchPersonBatchFn

  // Construct a data-loader for fetch follower requests
  let fetchFollowersBatchFn (requests : Set<string>) =
    async {
      let! responses =
        requests
        |> Seq.map (fun id ->
          async {
            let! followerIDs = DB.fetchFollowers id

            return id, followerIDs
          })
        |> Async.Parallel

      return Map.ofSeq responses
    }

  let fetchFollowersContext = DataLoader.context ()
  let fetchFollowersLoader = 
    DataLoader.create fetchFollowersContext fetchFollowersBatchFn

  let root =
    {
      FetchPerson = fun id -> fetchPersonLoader.GetAsync(id)
      FetchFollowers = fun id -> fetchFollowersLoader.GetAsync(id)
    }

  // Uncomment this to see how sub-optimal the query is without the data-loader
  // let root =
  //   {
  //     FetchPerson = DB.fetchPerson
  //     FetchFollowers = DB.fetchFollowers
  //   }

  // See https://bartoszsypytkowski.com/data-loaders/
  do! Async.SwitchToContext fetchPersonContext
  do! Async.SwitchToContext fetchFollowersContext

  // Execute the query
  let! response = executor.AsyncExecute(query, root)

  printfn "%A" response
}
|> Async.RunSynchronously