From 51ed1e2c7a0cf847338ece2ba80e5158e3287a27 Mon Sep 17 00:00:00 2001 From: Guntis Smaukstelis Date: Mon, 11 Dec 2023 00:12:52 +0200 Subject: [PATCH] Improving insert queries for postgres --- README.md | 7 +++++-- src/main/scala/db/DataService.scala | 11 +++++++++-- src/main/scala/db/PostgresService.scala | 22 ++++++++++++---------- src/main/scala/server/Server.scala | 1 + 4 files changed, 27 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index c10b5ef..20c5deb 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,7 @@ npm run dev ``` SELECT * FROM public.weather WHERE datetime > '2023-12-9T00:00:00+00'::timestamptz; DELETE FROM public.weather +SET TIMEZONE = 'Europe/Riga'; ``` ## Docker @@ -53,6 +54,8 @@ docker start local-postgres ## Fly Postgres ``` -fly machines list --app -fly machines start --app +fly machines list --app weather-tool-db +fly machines start --app weather-tool-db + +fly proxy 15432:5432 -a weather-tool-db // map to local port for pgAdmin ``` \ No newline at end of file diff --git a/src/main/scala/db/DataService.scala b/src/main/scala/db/DataService.scala index ff1a701..3340802 100644 --- a/src/main/scala/db/DataService.scala +++ b/src/main/scala/db/DataService.scala @@ -58,10 +58,17 @@ class DataService private( def save(fileName: String, content: String): IO[String] = { // TODO replace unsafeRunSync to redeemWith - postgresService.save(fileName, content).unsafeRunSync() +// postgresService.save(fileName, content).unsafeRunSync() +// postgresService.save(fileName, content) + + for { + result <- fileService.save(fileName, content) + _ <- postgresService.save(fileName, content) + } yield result // TODO delete this - fileService.save(fileName, content) +// fileService.save(fileName, content) + // fileService.save(fileName, content).redeemWith( // error => IO.raiseError(error), // savedFileName => { diff --git a/src/main/scala/db/PostgresService.scala b/src/main/scala/db/PostgresService.scala index 8ac3526..ee76f36 100644 --- a/src/main/scala/db/PostgresService.scala +++ b/src/main/scala/db/PostgresService.scala @@ -85,17 +85,19 @@ class PostgresService(transactor: Transactor[IO], log: Logger[IO]) extends DataS implicit val doubleOptionMeta: Meta[Option[Double]] = Meta[Double].imap(Option(_))(_.getOrElse(Double.NaN)) def insertInWeatherTable(data: List[WeatherStationData]): IO[Int] = { - data.parTraverse(line => { - val zonedTime: ZonedDateTime = line.timestamp.atZone(rigaZone) - val w = line.weather - for { - insertTableSql <- getResourceContent("/db/insert_weather_table.sql") - result <- Update[(ZonedDateTime, String, Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], List[String])]( - insertTableSql - ).run((zonedTime, line.city, w.tempMax, w.tempMin, w.tempAvg, w.precipitation, w.windAvg, w.windMax, w.visibilityMin, w.visibilityAvg, w.snowAvg, w.atmPressure, w.dewPoint, w.humidity, w.sunDuration, w.phenomena)) +// IO.blocking { + getResourceContent("/db/insert_weather_table.sql").flatMap { insertTableSql => + val insertData = data.map(line => { + val zonedTime: ZonedDateTime = line.timestamp.atZone(rigaZone) + val w = line.weather + (zonedTime, line.city, w.tempMax, w.tempMin, w.tempAvg, w.precipitation, w.windAvg, w.windMax, w.visibilityMin, w.visibilityAvg, w.snowAvg, w.atmPressure, w.dewPoint, w.humidity, w.sunDuration, w.phenomena) + }) + + Update[(ZonedDateTime, String, Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], List[String])](insertTableSql) + .updateMany(insertData) .transact(transactor) - } yield result - }).map(_.sum) + } +// }.flatten } def selectWeatherTable(): IO[List[(String, Option[Double])]] = { diff --git a/src/main/scala/server/Server.scala b/src/main/scala/server/Server.scala index 217d9ba..4a9369f 100644 --- a/src/main/scala/server/Server.scala +++ b/src/main/scala/server/Server.scala @@ -67,6 +67,7 @@ class Server(dataService: DataService, fetch: FetchService, log: Logger[IO]) { fetchServiceError = fetchResultEither.left.toOption.map(e => s"FetchServiceError: ${e.getMessage}").toList fetchResult = fetchResultEither.getOrElse(List.empty) (fetchErrors, successDownloads) = fetchResult.partitionMap(identity) + _ <- log.info(s"FETCHED SUCCESSFULLY files: ${successDownloads.size}") saveResults <- successDownloads.traverse { case (name, content) => dataService.save(name, content).attempt } (saveErrors, successSaves) = saveResults.partitionMap(identity) // successes = successDownloads.map(s => s"fetched: ${s._1}") ++ successSaves.map(s => s"saved: $s")