From 47be3bf38546a957bdd430fae604bb5f88d1328f Mon Sep 17 00:00:00 2001 From: Guntis Smaukstelis Date: Mon, 23 Oct 2023 16:20:45 +0300 Subject: [PATCH] DB insert/update rows from file, query dates from months --- src/main/scala/app/Main.scala | 16 ++++- src/main/scala/db/DataService.scala | 17 +++++- src/main/scala/db/PostgresService.scala | 79 +++++++++++++++++++------ 3 files changed, 89 insertions(+), 23 deletions(-) diff --git a/src/main/scala/app/Main.scala b/src/main/scala/app/Main.scala index 0007672..14a4ded 100644 --- a/src/main/scala/app/Main.scala +++ b/src/main/scala/app/Main.scala @@ -2,15 +2,27 @@ package app import cats.effect._ import cats.implicits.catsSyntaxTuple2Parallel -import db.{FileService, DataService} +import db.Main.transactor +import db.{DataService, FileService, PostgresService} +import doobie.Transactor import fetch.{FetchService, FileFetchScheduler} import server.Server object Main extends IOApp { + + def transactor[F[_] : Async]: Transactor[F] = Transactor.fromDriverManager[F]( + "org.postgresql.Driver", + "jdbc:postgresql://localhost:5432/weather-tool", + "postgres", + "mysecretpassword" + ) def run(args: List[String]): IO[ExitCode] = { + val xa = transactor[IO] + for { + postgresService <- PostgresService.of(xa) fileService <- FileService.of - dataService <- DataService.of(fileService) + dataService <- DataService.of(fileService, postgresService) fetch <- FetchService.of fileFetchScheduler <- FileFetchScheduler.of(dataService, fetch) diff --git a/src/main/scala/db/DataService.scala b/src/main/scala/db/DataService.scala index ae4f09e..1348d49 100644 --- a/src/main/scala/db/DataService.scala +++ b/src/main/scala/db/DataService.scala @@ -1,5 +1,6 @@ package db +import cats.effect.unsafe.implicits.global import cats.effect.{Clock, IO, Ref} import cats.implicits.toTraverseOps import fetch.FileNameService @@ -14,10 +15,12 @@ trait DataServiceTrait { def getInRange(from: LocalDateTime, to: LocalDateTime): IO[List[String]] def getDates: IO[List[LocalDate]] def getDateFileNames(date: LocalDate): IO[List[String]] + + def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] } object DataService { - def of(fileService: FileService): IO[DataService] = { + def of(fileService: FileService, postgresService: PostgresService): IO[DataService] = { for { log <- Slf4jLogger.create[IO] fileNameService = new FileNameService() @@ -26,11 +29,12 @@ object DataService { .map(content => (fileName, content))) state = contents.toMap stateRef <- Ref.of[IO, Map[String, List[String]]](state) - } yield new DataService(fileService, new FileNameService(), log, stateRef) + } yield new DataService(fileService, postgresService, new FileNameService(), log, stateRef) } } class DataService private( fileService: FileService, + postgresService: PostgresService, fileNameService: FileNameService, log: Logger[IO], private val state: Ref[IO, Map[String, List[String]]] @@ -47,6 +51,10 @@ class DataService private( } def save(fileName: String, content: String): IO[String] = { + // TODO remove unsafeRunSync + postgresService.save(fileName, content).unsafeRunSync() + + // TODO delete this fileService.save(fileName, content).redeemWith( error => IO.raiseError(error), savedFileName => { @@ -63,7 +71,10 @@ class DataService private( def getDates: IO[List[LocalDate]] = fileService.getDates - def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] = fileService.getDatesByMonths(monthList) + def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] = { +// fileService.getDatesByMonths(monthList) + postgresService.getDatesByMonths(monthList) + } def getDateFileNames(date: LocalDate): IO[List[String]] = fileService.getDateFileNames(date) diff --git a/src/main/scala/db/PostgresService.scala b/src/main/scala/db/PostgresService.scala index 1d022ad..3c66b29 100644 --- a/src/main/scala/db/PostgresService.scala +++ b/src/main/scala/db/PostgresService.scala @@ -2,14 +2,16 @@ package db import cats.data.NonEmptyList import cats.effect._ +import cats.implicits.{catsSyntaxParallelTraverse1, toFoldableOps} import doobie._ import doobie.implicits._ import java.time.format.DateTimeFormatter -import java.time.{LocalDateTime, ZoneId, ZonedDateTime} +import java.time.{LocalDate, LocalDateTime, ZoneId, ZonedDateTime} import doobie.postgres.implicits._ import org.typelevel.log4cats.Logger import org.typelevel.log4cats.slf4j.Slf4jLogger +import parse.{Parser, WeatherStationData} object PostgresService { def of(transactor: Transactor[IO]): IO[PostgresService] = { @@ -17,7 +19,52 @@ object PostgresService { } } -class PostgresService(transactor: Transactor[IO], log: Logger[IO]) { +class PostgresService(transactor: Transactor[IO], log: Logger[IO]) extends DataServiceTrait { + private val rigaZone = ZoneId.of("Europe/Riga") + // in pgAdmin run: ```SET TIMEZONE = 'Europe/Riga';``` + + def save(fileName: String, content: String): IO[String] = { + val strLines = content.split(System.lineSeparator()).toList + val weatherStationData = strLines.flatMap(Parser.parseLine) + insertInWeatherTable(weatherStationData) + .attempt + .flatMap { + case Left(error) => log.error(s"Write db '$fileName' failed with error: ${error.getMessage}") *> IO.raiseError(error) + case Right(rowCount) => log.info(s"write rows: $rowCount file: $fileName").as(fileName) + } + } + + def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] = { + val monthYearPairs = monthList.map { localDate => + val month = localDate.atStartOfDay.atZone(rigaZone).getMonthValue + val year = localDate.atStartOfDay.atZone(rigaZone).getYear + (month, year) + } + + val whereClauses = monthYearPairs.map { + case (month, year) => fr"(EXTRACT(MONTH FROM dateTime) = $month AND EXTRACT(YEAR FROM dateTime) = $year)" + } + + val combinedWhereClause = whereClauses.intercalate(fr" OR ") + + ( + fr""" + SELECT DISTINCT DATE(dateTime) + FROM weather + WHERE """ ++ combinedWhereClause + ) + .query[LocalDate] + .to[List] + .transact(transactor) + } + + def readFile(fileName: String): IO[List[String]] = ??? + + def getInRange(from: LocalDateTime, to: LocalDateTime): IO[List[String]] = ??? + + def getDates: IO[List[LocalDate]] = ??? + + def getDateFileNames(date: LocalDate): IO[List[String]] = ??? def getResourceContent(path: String): IO[String] = { val streamResource = Resource.make(IO(getClass.getResourceAsStream(path))) { stream => IO(stream.close()).handleErrorWith(_ => IO.unit) @@ -37,26 +84,22 @@ class PostgresService(transactor: Transactor[IO], log: Logger[IO]) { implicit val doubleOptionMeta: Meta[Option[Double]] = Meta[Double].imap(Option(_))(_.getOrElse(Double.NaN)) - def insertInWeatherTable(): IO[Int] = { - val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm") - val dateTime = LocalDateTime.parse("20230516_1500", formatter) - - val rigaZone = ZoneId.of("Europe/Riga") - val zonedTime: ZonedDateTime = dateTime.atZone(rigaZone) - - // in pgAdmin run: ```SET TIMEZONE = 'Europe/Riga';``` - - 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, "Rīga", Some(20.2), Some(8.8), Some(15.6), None, None, None, None, None, None, None, None, None, None, List("hail", "rain"))).transact(transactor) - } yield result + 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)) + .transact(transactor) + } yield result + }).map(_.sum) } def selectWeatherTable(): IO[List[(String, Option[Double])]] = { val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm") - val rigaZone = ZoneId.of("Europe/Riga") val from = LocalDateTime.parse("20230516_0100", formatter).atZone(rigaZone) val to = LocalDateTime.parse("20230516_1500", formatter).atZone(rigaZone) val citiesNel = NonEmptyList.of("Rīga", "Rēzekne")