如何使用 Akka HTTP 解组 json 响应删除不必要的字段

Posted 2023-03-23

【中文标题】如何使用 Akka HTTP 解组 json 响应删除不必要的字段

【英文标题】：How to unmarshall json response removing unnecessary fields using Akka HTTP

【发布时间】：2022-01-10 08:02:39

【问题描述】：

我是 Akka HTTP 的新手，我想从 JSON 响应中删除不必要的字段，只获取必要的字段。例如，我使用this 端点来获取响应，它包含一堆字段。目前我只需要“名称”和“版本”。我想知道如何将其反序列化为仅包含“名称”和“版本”的案例类。我编写了以下几行代码以将响应作为字符串获取。

import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpRequest, HttpResponse
import akka.stream.scaladsl.Flow, Sink, Source
import akka.stream.ActorMaterializer, OverflowStrategy

import scala.concurrent.Future
import scala.concurrent.duration.DurationInt
import scala.language.postfixOps
import scala.util.Failure, Success

object SoftwareRegistry extends App 

  implicit val system = ActorSystem("NPMRegistry")
  implicit val materializer = ActorMaterializer()

  import system.dispatcher

  case class NPMPackage(name: String)

  // reading the packages
  val filename = "B:\\Scala\\NPMRegistry\\src\\main\\resources\\packages.txt"
  val bufferedSource = scala.io.Source.fromFile(filename)
  val listOfPackages: List[NPMPackage] = (for (line <- bufferedSource.getLines) yield 
    NPMPackage(line.trim)
  ).toList
  bufferedSource.close()

  // source
  val sourceList = Source(listOfPackages)

  // sink
  val sink = Sink.foreach[NPMPackage]  p =>
    // https request
    val responseFuture: Future[HttpResponse] =
      Http().singleRequest(HttpRequest(uri = s"https://registry.npmjs.org/$p.name"))
    val x = responseFuture
      .flatMap(_.entity.toStrict(2 seconds))
      .map(_.data.utf8String)
    x.onComplete 
      case Success(res) => println(res)
      case Failure(_) => sys.error("Something went wrong")
    
  

  // flow to slow things down and streaming sink to time-delayed operations
  val bufferedFlow = Flow[NPMPackage]
    .buffer(10, overflowStrategy = OverflowStrategy.backpressure)
    .throttle(1, 3 seconds)

  sourceList.async
    .via(bufferedFlow).async
    .to(sink)
    .run()

它会打印以下输出

【问题讨论】：

预期的数据类型是什么？在这种特殊情况下，“版本”本身就是一个对象，您需要该对象提供什么？

@JohnyTKoshy 我需要每个版本的 'dependencies' 和 'devDependencies'

啊.. 这些又是对象。我发布了一个答案，其版本为List。其余的你也许能弄清楚。

@JohnyTKoshy 这似乎有效，我试图找出嵌套对象类型，这让我有点困惑。我创建了以下数据结构。 case class Package(name: String, versions: List[Version]) case class Version(version: String, dependencies: List[String], devDependencies: List[String])。你介意帮我按照这些数据结构写吗？

dependencies 和 devDependencies 不是字符串，它们是 javascript 对象。

【参考方案1】：

为了解析json，你需要使用一些库。在akka-http 文档中，他们使用spray-json。使用适当的akkaHttpVersion 将以下依赖项添加到您的build.sbt。

"com.typesafe.akka" %% "akka-http-spray-json" % akkaHttpVersion

现在您的数据需要序列化器和反序列化器。我正在使用一个简单的模型，根据需要更改它。

trait Formatter extends DefaultJsonProtocol 

  implicit object jsonFormat extends JsonFormat[Versions] 
    override def read(json: JsValue): Versions = json match 
      case JsObject(fields) =>
        Versions(fields.keys.toList)
    

    override def write(obj: Versions): JsValue = JsonParser(obj.toString)
  

  implicit val formatterPackage: RootJsonFormat[Package] = jsonFormat2(Package)

  case class Package(name: String, versions: Versions)

  case class Versions(versions: List[String])

终于sink:

 //needed import with others
 import spray.json._

 object SoftwareRegistry extends App  with Formatter 

   //existing code
   //---------


   val sink = Sink.foreach[NPMPackage]  p =>
       // https request
       val responseFuture: Future[HttpResponse] =
         Http().singleRequest(HttpRequest(uri = s"https://registry.npmjs.org/$p.name"))
       val packages = responseFuture
         .flatMap(
           _.entity
             .dataBytes
             .via(JsonFraming.objectScanner(Int.MaxValue))
             .map(_.utf8String)
             .map(_.parseJson.convertTo[Package])
             .toMat(Sink.seq)(Keep.right)
             .run()
         )

       packages.onComplete 
         case Success(res) => println(res)
         case Failure(_) => sys.error("Something went wrong")
       
   

   //existing code
   //---------