DB insert/update rows from file, query dates from months

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)

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)
 

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] = {
+  def insertInWeatherTable(data: List[WeatherStationData]): IO[Int] = {
-    val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
+    data.parTraverse(line => {
-    val dateTime = LocalDateTime.parse("20230516_1500", formatter)
+      val zonedTime: ZonedDateTime = line.timestamp.atZone(rigaZone)
-
+      val w = line.weather
-    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)
+        ).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")