Refactor to FileService and PostgresService
This commit is contained in:
@@ -2,15 +2,15 @@ package app
|
||||
|
||||
import cats.effect._
|
||||
import cats.implicits.catsSyntaxTuple2Parallel
|
||||
import db.{DBService, DataService}
|
||||
import db.{FileService, DataService}
|
||||
import fetch.{FetchService, FileFetchScheduler}
|
||||
import server.Server
|
||||
|
||||
object Main extends IOApp {
|
||||
def run(args: List[String]): IO[ExitCode] = {
|
||||
for {
|
||||
dbService <- DBService.of
|
||||
dataService <- DataService.of(dbService)
|
||||
fileService <- FileService.of
|
||||
dataService <- DataService.of(fileService)
|
||||
|
||||
fetch <- FetchService.of
|
||||
fileFetchScheduler <- FileFetchScheduler.of(dataService, fetch)
|
||||
|
||||
@@ -17,20 +17,20 @@ trait DataServiceTrait {
|
||||
}
|
||||
|
||||
object DataService {
|
||||
def of(dbService: DBService): IO[DataService] = {
|
||||
def of(fileService: FileService): IO[DataService] = {
|
||||
for {
|
||||
log <- Slf4jLogger.create[IO]
|
||||
fileNameService = new FileNameService()
|
||||
fileNames <- fileNameService.generateLast24Hours
|
||||
contents <- fileNames.traverse(fileName => dbService.readFile(fileName)
|
||||
contents <- fileNames.traverse(fileName => fileService.readFile(fileName)
|
||||
.map(content => (fileName, content)))
|
||||
state = contents.toMap
|
||||
stateRef <- Ref.of[IO, Map[String, List[String]]](state)
|
||||
} yield new DataService(dbService, new FileNameService(), log, stateRef)
|
||||
} yield new DataService(fileService, new FileNameService(), log, stateRef)
|
||||
}
|
||||
}
|
||||
class DataService private(
|
||||
dbService: DBService,
|
||||
fileService: FileService,
|
||||
fileNameService: FileNameService,
|
||||
log: Logger[IO],
|
||||
private val state: Ref[IO, Map[String, List[String]]]
|
||||
@@ -47,7 +47,7 @@ class DataService private(
|
||||
}
|
||||
|
||||
def save(fileName: String, content: String): IO[String] = {
|
||||
dbService.save(fileName, content).redeemWith(
|
||||
fileService.save(fileName, content).redeemWith(
|
||||
error => IO.raiseError(error),
|
||||
savedFileName => {
|
||||
state.update(st => st.updated(savedFileName, content.split("\n").toList)) *>
|
||||
@@ -57,15 +57,15 @@ class DataService private(
|
||||
).onError(error => log.info(s"errr... $error"))
|
||||
}
|
||||
|
||||
def readFile(fileName: String): IO[List[String]] = dbService.readFile(fileName)
|
||||
def readFile(fileName: String): IO[List[String]] = fileService.readFile(fileName)
|
||||
|
||||
def getInRange(from: LocalDateTime, to: LocalDateTime): IO[List[String]] = dbService.getInRange(from, to)
|
||||
def getInRange(from: LocalDateTime, to: LocalDateTime): IO[List[String]] = fileService.getInRange(from, to)
|
||||
|
||||
def getDates: IO[List[LocalDate]] = dbService.getDates
|
||||
def getDates: IO[List[LocalDate]] = fileService.getDates
|
||||
|
||||
def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] = dbService.getDatesByMonths(monthList)
|
||||
def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] = fileService.getDatesByMonths(monthList)
|
||||
|
||||
def getDateFileNames(date: LocalDate): IO[List[String]] = dbService.getDateFileNames(date)
|
||||
def getDateFileNames(date: LocalDate): IO[List[String]] = fileService.getDateFileNames(date)
|
||||
|
||||
// TODO implement getting full data from state
|
||||
def getLast24Hours: IO[List[String]] = {
|
||||
|
||||
@@ -14,13 +14,13 @@ import scala.concurrent.ExecutionContext
|
||||
import scala.io.Source
|
||||
import scala.util.Try
|
||||
|
||||
object DBService {
|
||||
def of: IO[DBService] = {
|
||||
Slf4jLogger.create[IO].map(logger => new DBService(logger))
|
||||
object FileService {
|
||||
def of: IO[FileService] = {
|
||||
Slf4jLogger.create[IO].map(logger => new FileService(logger))
|
||||
}
|
||||
}
|
||||
|
||||
class DBService(log: Logger[IO]) extends DataServiceTrait {
|
||||
class FileService(log: Logger[IO]) extends DataServiceTrait {
|
||||
private val dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
|
||||
private val dataPath = "./data"
|
||||
private val nonDuplicatedLines = 34 // takes only first 34 lines of data as rest after 'Zosēni' is duplicated
|
||||
@@ -0,0 +1,32 @@
|
||||
package db
|
||||
|
||||
import cats.effect._
|
||||
import cats.effect.unsafe.implicits.global
|
||||
import doobie._
|
||||
import doobie.implicits._
|
||||
import doobie.postgres.implicits._
|
||||
|
||||
object Main {
|
||||
def transactor[F[_]: Async]: Transactor[F] = Transactor.fromDriverManager[F](
|
||||
"org.postgresql.Driver",
|
||||
"jdbc:postgresql://localhost:5432/weather-tool",
|
||||
"postgres",
|
||||
"mysecretpassword"
|
||||
)
|
||||
|
||||
def main(args: Array[String]): Unit = {
|
||||
val xa = transactor[IO]
|
||||
val result = for {
|
||||
postgresService <- PostgresService.of(xa)
|
||||
re <- postgresService.selectWeatherTable
|
||||
} yield re
|
||||
|
||||
|
||||
// val result = createWeatherTable(xa).unsafeRunSync()
|
||||
// val result = insertInWeatherTable(xa).unsafeRunSync()
|
||||
// val result = selectWeatherTable(xa).unsafeRunSync()
|
||||
// val result = dropWeatherTable(xa).unsafeRunSync()
|
||||
|
||||
println(s"result: ${result.unsafeRunSync()}")
|
||||
}
|
||||
}
|
||||
@@ -2,24 +2,22 @@ package db
|
||||
|
||||
import cats.data.NonEmptyList
|
||||
import cats.effect._
|
||||
import cats.effect.unsafe.implicits.global
|
||||
import doobie._
|
||||
import doobie.implicits._
|
||||
|
||||
import java.time.format.DateTimeFormatter
|
||||
import java.time.{LocalDateTime, ZoneId, ZonedDateTime}
|
||||
import doobie.postgres.implicits._
|
||||
import org.typelevel.log4cats.Logger
|
||||
import org.typelevel.log4cats.slf4j.Slf4jLogger
|
||||
|
||||
case class SqlContent(value: String)
|
||||
|
||||
object Postgres {
|
||||
def transactor[F[_]: Async]: Transactor[F] = Transactor.fromDriverManager[F](
|
||||
"org.postgresql.Driver",
|
||||
"jdbc:postgresql://localhost:5432/weather-tool",
|
||||
"postgres",
|
||||
"mysecretpassword"
|
||||
)
|
||||
object PostgresService {
|
||||
def of(transactor: Transactor[IO]): IO[PostgresService] = {
|
||||
Slf4jLogger.create[IO].map(logger => new PostgresService(transactor, logger))
|
||||
}
|
||||
}
|
||||
|
||||
class PostgresService(transactor: Transactor[IO], log: Logger[IO]) {
|
||||
def getResourceContent(path: String): IO[String] = {
|
||||
val streamResource = Resource.make(IO(getClass.getResourceAsStream(path))) { stream =>
|
||||
IO(stream.close()).handleErrorWith(_ => IO.unit)
|
||||
@@ -30,23 +28,16 @@ object Postgres {
|
||||
}
|
||||
}
|
||||
|
||||
// def findUserById(userId: Int)(implicit xa: Transactor[IO]): IO[Option[User]] = {
|
||||
// sql"SELECT id, name FROM users WHERE id = $userId"
|
||||
// .query[User]
|
||||
// .option
|
||||
// .transact(xa)
|
||||
// }
|
||||
|
||||
def createWeatherTable(implicit xa: Transactor[IO]): IO[Int] = {
|
||||
def createWeatherTable(): IO[Int] = {
|
||||
for {
|
||||
createTableSql <- getResourceContent("/db/create_weather_table.sql")
|
||||
result <- Update0(createTableSql, None).run.transact(xa)
|
||||
result <- Update0(createTableSql, None).run.transact(transactor)
|
||||
} yield result
|
||||
}
|
||||
|
||||
implicit val doubleOptionMeta: Meta[Option[Double]] = Meta[Double].imap(Option(_))(_.getOrElse(Double.NaN))
|
||||
|
||||
def insertInWeatherTable(xa: Transactor[IO]): IO[Int] = {
|
||||
def insertInWeatherTable(): IO[Int] = {
|
||||
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
|
||||
val dateTime = LocalDateTime.parse("20230516_1500", formatter)
|
||||
|
||||
@@ -59,11 +50,11 @@ object Postgres {
|
||||
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(xa)
|
||||
).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 selectWeatherTable(xa: Transactor[IO]): IO[List[(String, Option[Double])]] = {
|
||||
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)
|
||||
@@ -76,8 +67,7 @@ object Postgres {
|
||||
fr"""
|
||||
SELECT
|
||||
city,
|
||||
MIN(""" ++ Fragment.const(columnName) ++
|
||||
fr""") AS tempMin
|
||||
MIN(""" ++ Fragment.const(columnName) ++ fr""") AS tempMin
|
||||
FROM weather
|
||||
WHERE
|
||||
""" ++ Fragments.in(fr"city", citiesNel) ++ fr"""
|
||||
@@ -89,24 +79,13 @@ object Postgres {
|
||||
baseQuery
|
||||
.query[(String, Option[Double])]
|
||||
.to[List]
|
||||
.transact(xa)
|
||||
.transact(transactor)
|
||||
}
|
||||
|
||||
def dropWeatherTable(implicit xa: Transactor[IO]): IO[Int] = {
|
||||
def dropWeatherTable(): IO[Int] = {
|
||||
for {
|
||||
dropTableSql <- getResourceContent("/db/drop_weather_table.sql")
|
||||
result <- Update0(dropTableSql, None).run.transact(xa)
|
||||
result <- Update0(dropTableSql, None).run.transact(transactor)
|
||||
} yield result
|
||||
}
|
||||
|
||||
def main(args: Array[String]): Unit = {
|
||||
val xa = transactor[IO]
|
||||
|
||||
// val result = createWeatherTable(xa).unsafeRunSync()
|
||||
// val result = insertInWeatherTable(xa).unsafeRunSync()
|
||||
val result = selectWeatherTable(xa).unsafeRunSync()
|
||||
// val result = dropWeatherTable(xa).unsafeRunSync()
|
||||
|
||||
println(s"result: $result")
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
package parse
|
||||
|
||||
import cats.effect.unsafe.implicits.global
|
||||
import db.DBService
|
||||
import db.FileService
|
||||
import io.circe.syntax.EncoderOps
|
||||
import parse.Aggregate.{AggregateKey, UserQuery}
|
||||
|
||||
@@ -43,8 +43,8 @@ object Main {
|
||||
val from = LocalDateTime.parse("20230627_0000", formatter)
|
||||
val to = LocalDateTime.parse("20230627_2359", formatter)
|
||||
|
||||
val dbService = DBService.of.unsafeRunSync()
|
||||
val lines = dbService.getInRange(from, to).unsafeRunSync()
|
||||
val fileService = FileService.of.unsafeRunSync()
|
||||
val lines = fileService.getInRange(from, to).unsafeRunSync()
|
||||
val query = UserQuery(List("Ainaži"), "tempAvg", AggregateKey.Avg, ChronoUnit.HOURS)
|
||||
val parsed = Parser.queryData(query, lines)
|
||||
println(parsed.asJson)
|
||||
|
||||
@@ -9,12 +9,12 @@ import java.time.format.DateTimeFormatter
|
||||
import java.time.{LocalDate, LocalDateTime}
|
||||
|
||||
|
||||
class DBServiceSpec extends AnyFunSuite with Matchers {
|
||||
class FileServiceSpec extends AnyFunSuite with Matchers {
|
||||
private val dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
|
||||
|
||||
test("getDateFileNames should return correct file names") {
|
||||
val dbService = DBService.of.unsafeRunSync()
|
||||
val datesList = dbService.getDateFileNames(LocalDate.of(2023, 5, 15))
|
||||
val fileService = FileService.of.unsafeRunSync()
|
||||
val datesList = fileService.getDateFileNames(LocalDate.of(2023, 5, 15))
|
||||
.unsafeRunSync()
|
||||
|
||||
val expectedFileNames = (0 to 23).toList
|
||||
@@ -28,8 +28,8 @@ class DBServiceSpec extends AnyFunSuite with Matchers {
|
||||
val from = LocalDateTime.parse("20230513_2200", dateFormatter)
|
||||
val to = LocalDateTime.parse("20230516_1230", dateFormatter)
|
||||
|
||||
val dbService = DBService.of.unsafeRunSync()
|
||||
val lines = dbService.getInRange(from, to).unsafeRunSync()
|
||||
val fileService = FileService.of.unsafeRunSync()
|
||||
val lines = fileService.getInRange(from, to).unsafeRunSync()
|
||||
|
||||
lines.length shouldBe 2142
|
||||
}
|
||||
@@ -41,8 +41,8 @@ class DBServiceSpec extends AnyFunSuite with Matchers {
|
||||
LocalDate.parse("20230501", monthFormatter),
|
||||
LocalDate.parse("20230601", monthFormatter),
|
||||
)
|
||||
val dbService = DBService.of.unsafeRunSync()
|
||||
val dates = dbService.getDatesByMonths(monthList).unsafeRunSync()
|
||||
val fileService = FileService.of.unsafeRunSync()
|
||||
val dates = fileService.getDatesByMonths(monthList).unsafeRunSync()
|
||||
|
||||
dates should not be empty
|
||||
|
||||
@@ -56,19 +56,19 @@ class DBServiceSpec extends AnyFunSuite with Matchers {
|
||||
}
|
||||
|
||||
test("DBService.save should return correct result") {
|
||||
val dbService = DBService.of.unsafeRunSync()
|
||||
val fileService = FileService.of.unsafeRunSync()
|
||||
val fileName = "testFile.txt"
|
||||
val fileContent = "test content..."
|
||||
|
||||
dbService.save(fileName, fileContent).unsafeRunSync() shouldEqual fileName
|
||||
fileService.save(fileName, fileContent).unsafeRunSync() shouldEqual fileName
|
||||
}
|
||||
|
||||
test("DBService.save returns error on invalid file name") {
|
||||
val dbService = DBService.of.unsafeRunSync()
|
||||
val fileService = FileService.of.unsafeRunSync()
|
||||
val fileName = "/invalid/file/name"
|
||||
val fileContent = "test content..."
|
||||
|
||||
val result = dbService.save(fileName, fileContent).attempt.unsafeRunSync()
|
||||
val result = fileService.save(fileName, fileContent).attempt.unsafeRunSync()
|
||||
|
||||
result match {
|
||||
case Left(e) =>
|
||||
@@ -3,7 +3,7 @@ package fetch
|
||||
import cats.effect.IO
|
||||
import cats.effect.unsafe.implicits.global
|
||||
import cats.implicits.toTraverseOps
|
||||
import db.DBService
|
||||
import db.FileService
|
||||
import org.scalatest.funsuite.AnyFunSuite
|
||||
import org.scalatest.matchers.should.Matchers
|
||||
import org.typelevel.log4cats.slf4j.Slf4jLogger
|
||||
|
||||
@@ -4,7 +4,7 @@ import base.IOSuite
|
||||
import cats.effect.{Clock, IO}
|
||||
import cats.effect.kernel.Ref
|
||||
import cats.implicits.catsSyntaxTuple3Semigroupal
|
||||
import db.DBService
|
||||
import db.FileService
|
||||
import org.scalatest.matchers.should.Matchers
|
||||
import org.scalatest.wordspec.AsyncWordSpec
|
||||
import org.typelevel.log4cats.slf4j.Slf4jLogger
|
||||
@@ -19,7 +19,7 @@ class FileFetchSchedulerSpec extends AsyncWordSpec with Matchers with IOSuite {
|
||||
refFetch <- Ref.of[IO, Option[(String, String)]](None)
|
||||
refScheduler <- Ref.of[IO, Option[Either[Throwable, (String, String)]]](None)
|
||||
log <- Slf4jLogger.create[IO]
|
||||
dbService = new DBService(log) {
|
||||
fileService = new FileService(log) {
|
||||
override def save(fileName: String, content: String): IO[String] = {
|
||||
refDb.set(Some(fileName)).as(fileName)
|
||||
}
|
||||
@@ -40,7 +40,7 @@ class FileFetchSchedulerSpec extends AsyncWordSpec with Matchers with IOSuite {
|
||||
}
|
||||
}
|
||||
|
||||
res = new FileFetchScheduler(dbService, fetchService, fileNameService, scheduler, log)
|
||||
res = new FileFetchScheduler(fileService, fetchService, fileNameService, scheduler, log)
|
||||
.run.compile.drain *>
|
||||
(refDb.get, refFetch.get, refScheduler.get).tupled.map { case (db, fetch, scheduler) =>
|
||||
db shouldBe Some("file_test")
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
package parse
|
||||
|
||||
import cats.effect.unsafe.implicits.global
|
||||
import db.DBService
|
||||
import db.FileService
|
||||
import org.scalatest.funsuite.AnyFunSuite
|
||||
import org.scalatest.matchers.should.Matchers
|
||||
import parse.Aggregate.{AggregateKey, DoubleValue, TimeDoubleList, UserQuery}
|
||||
@@ -19,9 +19,9 @@ class ParserSpec extends AnyFunSuite with Matchers {
|
||||
val to = LocalDateTime.parse("20230516_0942", formatter)
|
||||
val userQuery = UserQuery(List("Bauska", "Dagda", "Daugavgrīva", "Rīga"), "precipitation", AggregateKey.Sum, ChronoUnit.HOURS)
|
||||
|
||||
val dbService = DBService.of.unsafeRunSync()
|
||||
val fileService = FileService.of.unsafeRunSync()
|
||||
|
||||
val lines = dbService.getInRange(from, to).unsafeRunSync()
|
||||
val lines = fileService.getInRange(from, to).unsafeRunSync()
|
||||
val parsed = Parser.queryData(userQuery, lines)
|
||||
|
||||
parsed shouldBe HashMap(
|
||||
@@ -38,9 +38,9 @@ class ParserSpec extends AnyFunSuite with Matchers {
|
||||
val to = LocalDateTime.parse("20230516_0800", formatter)
|
||||
val userQuery = UserQuery(List("Rīga"), "precipitation", AggregateKey.List, ChronoUnit.HOURS)
|
||||
|
||||
val dbService = DBService.of.unsafeRunSync()
|
||||
val fileService = FileService.of.unsafeRunSync()
|
||||
|
||||
val lines = dbService.getInRange(from, to).unsafeRunSync()
|
||||
val lines = fileService.getInRange(from, to).unsafeRunSync()
|
||||
val parsed = Parser.queryData(userQuery, lines)
|
||||
|
||||
parsed shouldBe HashMap(
|
||||
|
||||
Reference in New Issue
Block a user