ZIO LMDB feeding with french town postal codes / published by https://github.com/dacr/code-examples-manager #55574d76-9b22-422c-abde-9869596ca25f/bbf6fc4e9904e2f513f39ef2457089c7b331191

zio-lmdb-feed-postal-codes.sc

// summary : ZIO LMDB feeding with french town postal codes
// keywords : scala, zio, lmdb, @testable
// publish : gist
// authors : David Crosson
// license : Apache NON-AI License Version 2.0 (https://raw.githubusercontent.com/non-ai-licenses/non-ai-licenses/main/NON-AI-APACHE2)
// id : 55574d76-9b22-422c-abde-9869596ca25f
// created-on : 2022-08-09T20:01:37+02:00
// managed-by : https://github.com/dacr/code-examples-manager
// run-with : scala-cli $file

// ---------------------
//> using scala  "3.3.1"
//> using dep "fr.janalyse::zio-lmdb:1.8.0"
//> using dep "com.softwaremill.sttp.client3::zio:3.9.2"
//> using javaOpt "--add-opens", "java.base/java.nio=ALL-UNNAMED", "--add-opens", "java.base/sun.nio.ch=ALL-UNNAMED"
// ---------------------

import zio.*
import zio.lmdb.*
import zio.json.*
import java.io.File
import java.util.UUID
import java.time.OffsetDateTime

import sttp.client3.*
import sttp.model.*
import sttp.capabilities.zio.ZioStreams
import sttp.client3.httpclient.zio.*
import zio.stream.*
import zio.stream.ZPipeline.{splitLines, utf8Decode}

object FeedWithOpenDataPostalCodes extends ZIOAppDefault {

  val openDataPostalCodesHome          = "https://www.data.gouv.fr/fr/datasets/base-officielle-des-codes-postaux/"
  val openDataPostalCodesDataSourceURI = "https://www.data.gouv.fr/fr/datasets/r/554590ab-ae62-40ac-8353-ee75162c05ee"

// ---------------------------------------------------------------------------------------------------------------------
  case class GPS(
    latitude: Double,
    longitude: Double
  ) derives JsonCodec

  case class PostalCode(
    townCode: String,
    townName: String,
    postalCode: String,
    secondaryTownName: Option[String],
    deliveryLabel: String,
    gps: Option[GPS]
  ) derives JsonCodec {
    val countyCode = townCode.take(if (townCode.startsWith("97")) 3 else 2)
  }

// ---------------------------------------------------------------------------------------------------------------------

  def stringToGPS(input: String): Option[GPS] = {
    input.split(",").map(_.trim) match {
      case Array(latitude, longitude) =>
        for {
          lat <- latitude.toDoubleOption
          lon <- longitude.toDoubleOption
        } yield GPS(lat, lon)
      case _                          => None
    }
  }

  def stringToPostalCode(input: String): Option[PostalCode] = {
    input.trim.split(";") match {
      case Array(townCode, townName, postalCode, "", deliveryLabel, position)                =>
        Option(PostalCode(townCode, townName, postalCode, None, deliveryLabel, stringToGPS(position)))
      case Array(townCode, townName, postalCode, secondaryTownName, deliveryLabel, position) =>
        Option(PostalCode(townCode, townName, postalCode, Some(secondaryTownName), deliveryLabel, stringToGPS(position)))
      case Array(townCode, townName, postalCode, "", deliveryLabel)                          =>
        Option(PostalCode(townCode, townName, postalCode, None, deliveryLabel, None))
      case Array(townCode, townName, postalCode, secondaryTownName, deliveryLabel)           =>
        Option(PostalCode(townCode, townName, postalCode, Some(secondaryTownName), deliveryLabel, None))
      case data                                                                              =>
        println("Unmanaged input : " + data.mkString(";"))
        None
    }
  }

  val postalCodesStreamLogic = for {
    backend       <- ZIO.service[SttpBackend[Task, ZioStreams]]
    dataSourceURI <- ZIO.from(Uri.parse(openDataPostalCodesDataSourceURI))
    request        = basicRequest
                       .get(dataSourceURI)
                       .response(asStreamUnsafe(ZioStreams))
                       .readTimeout(Duration.Infinity.asScala)
    response      <- backend.send(request)
    dataStream    <- ZIO.from(response.body)
  } yield dataStream.via(utf8Decode >>> splitLines).map(line => stringToPostalCode(line))

  def feedLogic(postalCodesCollection: LMDBCollection[PostalCode]) =
    for {
      postalCodesStream <- postalCodesStreamLogic
      _                 <- postalCodesStream.filter(_.isDefined).runForeach(somePostalCode => postalCodesCollection.upsertOverwrite(somePostalCode.get.postalCode, somePostalCode.get))
    } yield ()

  val collectionName = "postal-codes"

  val storeFranceTownPostalCodes = for {
    postalCodesCollection <- LMDB.collectionCreate[PostalCode](collectionName, failIfExists=false)
    storedSizeBefore      <- postalCodesCollection.size()
    _                     <- if (storedSizeBefore == 0L) feedLogic(postalCodesCollection).unit else ZIO.unit
    // _                     <- feedLogic(postalCodesCollection).unit
    storedSize            <- postalCodesCollection.size()
    postalCodes           <- postalCodesCollection.collect()
    _                     <- Console.printLine(s"collection $collectionName contains ${postalCodes.size} records")
    _                     <- ZIO.foreach(postalCodes)(record => Console.printLine(record))
    _                     <- Console.printLine(s"""database contains $storedSize postal codes""")
    lmdb                  <- ZIO.service[LMDB]
    _                     <- Console.printLine("""LMDB standard tools can be used to manage the database content : sudo apt-get install lmdb-utils""")
    _                     <- Console.printLine(s"""To get some statistics     : mdb_stat -s $collectionName ${lmdb.databasePath}/""")
    _                     <- Console.printLine(s"""To dump collection content : mdb_dump -p -s $collectionName ${lmdb.databasePath}/""")
  } yield ()

  override def run = storeFranceTownPostalCodes.provide(LMDB.liveWithDatabaseName("lmdb-data-simple-example"), Scope.default, HttpClientZioBackend.layer())

}

FeedWithOpenDataPostalCodes.main(Array.empty)