Server fetches and saves files. Also proper error handling
This commit is contained in:
@@ -6,7 +6,7 @@ import cats.implicits.toTraverseOps
|
|||||||
|
|
||||||
import java.io.File
|
import java.io.File
|
||||||
import java.nio.file.{Files, Paths}
|
import java.nio.file.{Files, Paths}
|
||||||
import java.time.LocalDateTime
|
import java.time.{LocalDate, LocalDateTime}
|
||||||
import java.time.format.DateTimeFormatter
|
import java.time.format.DateTimeFormatter
|
||||||
import scala.io.Source
|
import scala.io.Source
|
||||||
import scala.util.Try
|
import scala.util.Try
|
||||||
@@ -46,19 +46,35 @@ object DBService {
|
|||||||
} yield fileLines.flatten
|
} yield fileLines.flatten
|
||||||
}
|
}
|
||||||
|
|
||||||
def save(fileName: String, content: String): IO[Unit] = {
|
def save(fileName: String, content: String): IO[Either[Throwable, String]] = {
|
||||||
val path = Paths.get(s"$dataPath/$fileName")
|
val path = Paths.get(s"$dataPath/$fileName")
|
||||||
// TODO redeemWith instead of flatMap
|
IO(Files.writeString(path, content))
|
||||||
IO(Files.writeString(path, content)).attempt.flatMap {
|
.redeemWith(
|
||||||
case Right(_) => IO.println(s"write: $fileName")
|
error => IO(Left(error))
|
||||||
case Left(error) => IO.println(s"Write file '$fileName' failed with error: ${error.getMessage}")
|
// .flatTap(_ => IO.println(s"Write file '$fileName' failed with error: ${error.getMessage}"))
|
||||||
|
,
|
||||||
|
_ => IO(Right(fileName))
|
||||||
|
// .flatTap(_ => IO.println(s"write: $fileName"))
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def getDates(): IO[Set[LocalDate]] = {
|
||||||
|
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd")
|
||||||
|
for {
|
||||||
|
fileNames <- readFileNames(dataPath)
|
||||||
|
datesStr <- IO.pure(fileNames.map(_.take(8)).toSet)
|
||||||
|
dates <- IO.pure(datesStr.flatMap(str => {
|
||||||
|
Try(LocalDate.parse(str, formatter)).toOption
|
||||||
|
}))
|
||||||
|
} yield dates
|
||||||
|
}
|
||||||
|
|
||||||
|
def fetchDates(date: LocalDate): IO[List[String]] = {
|
||||||
|
???
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO remove this. Just testing
|
// TODO remove this. Just testing
|
||||||
private def run: IO[Unit] = {
|
private def testGetInRange: IO[Unit] = {
|
||||||
println("----------------> db main")
|
|
||||||
|
|
||||||
val from = LocalDateTime.parse("20230414_2200", dateFormatter)
|
val from = LocalDateTime.parse("20230414_2200", dateFormatter)
|
||||||
val to = LocalDateTime.parse("20230501_1230", dateFormatter)
|
val to = LocalDateTime.parse("20230501_1230", dateFormatter)
|
||||||
|
|
||||||
@@ -68,7 +84,16 @@ object DBService {
|
|||||||
} yield ()
|
} yield ()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private def testGetDates: IO[Unit] = {
|
||||||
|
for {
|
||||||
|
dates <- getDates()
|
||||||
|
_ <- IO.println(dates)
|
||||||
|
} yield ()
|
||||||
|
}
|
||||||
|
|
||||||
def main(args: Array[String]): Unit = {
|
def main(args: Array[String]): Unit = {
|
||||||
run.unsafeRunSync()
|
println("----------------> db main")
|
||||||
|
// testGetInRange.unsafeRunSync()
|
||||||
|
testGetDates.unsafeRunSync()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,42 +0,0 @@
|
|||||||
package fetch
|
|
||||||
|
|
||||||
import cats.effect._
|
|
||||||
import cats.implicits._
|
|
||||||
import org.http4s._
|
|
||||||
import org.http4s.blaze.client.BlazeClientBuilder
|
|
||||||
import org.http4s.client.Client
|
|
||||||
import org.http4s.headers.Authorization
|
|
||||||
|
|
||||||
import com.typesafe.config.ConfigFactory
|
|
||||||
|
|
||||||
import java.nio.file.{Files, Paths}
|
|
||||||
import scala.concurrent.ExecutionContext.global
|
|
||||||
|
|
||||||
object FetchData extends IOApp.Simple {
|
|
||||||
// TODO wrap config in IO
|
|
||||||
private val config = ConfigFactory.load()
|
|
||||||
private val basicCredentials = BasicCredentials(config.getString("username"), config.getString("password"))
|
|
||||||
private val baseUrl = Uri.unsafeFromString(config.getString("url")) // 20220831_1330.csv
|
|
||||||
|
|
||||||
def makeRequest(client: Client[IO], url: Uri): IO[Unit] = {
|
|
||||||
val fileName = url.path.toString()
|
|
||||||
for {
|
|
||||||
request <- Request[IO](Method.GET, url)
|
|
||||||
.withHeaders(Authorization(basicCredentials))
|
|
||||||
.pure[IO]
|
|
||||||
responseOrError <- client.expect[String](request).attempt
|
|
||||||
_ <- responseOrError match {
|
|
||||||
case Right(response) => db.DBService.save(fileName, response)
|
|
||||||
case Left(error) => IO(println(s"Request failed to url: $url with error: ${error.getMessage}"))
|
|
||||||
}
|
|
||||||
} yield ()
|
|
||||||
}
|
|
||||||
|
|
||||||
def run: IO[Unit] = {
|
|
||||||
val fileNames = FileName.generateLastNHours(10)
|
|
||||||
val urls = fileNames.map(baseUrl / _)
|
|
||||||
BlazeClientBuilder[IO](global).resource.use { client =>
|
|
||||||
urls.traverse(url => makeRequest(client, url)) // urls.map(...).sequence
|
|
||||||
}
|
|
||||||
}.as(ExitCode.Success)
|
|
||||||
}
|
|
||||||
@@ -0,0 +1,52 @@
|
|||||||
|
package fetch
|
||||||
|
|
||||||
|
import cats.effect._
|
||||||
|
import cats.implicits._
|
||||||
|
import org.http4s._
|
||||||
|
import org.http4s.blaze.client.BlazeClientBuilder
|
||||||
|
import org.http4s.client.Client
|
||||||
|
import org.http4s.headers.Authorization
|
||||||
|
import com.typesafe.config.ConfigFactory
|
||||||
|
|
||||||
|
import java.time.{LocalDate, LocalDateTime}
|
||||||
|
import scala.concurrent.ExecutionContext.global
|
||||||
|
|
||||||
|
object FetchService {
|
||||||
|
// TODO wrap config in IO
|
||||||
|
private val config = ConfigFactory.load()
|
||||||
|
private val basicCredentials = BasicCredentials(config.getString("username"), config.getString("password"))
|
||||||
|
private val baseUrl = Uri.unsafeFromString(config.getString("url")) // 20220831_1330.csv
|
||||||
|
|
||||||
|
private def makeRequest(client: Client[IO], url: Uri): IO[Either[Throwable, (String, String)]] = {
|
||||||
|
val fileName = url.path.toString()
|
||||||
|
for {
|
||||||
|
request <- Request[IO](Method.GET, url)
|
||||||
|
.withHeaders(Authorization(basicCredentials))
|
||||||
|
.pure[IO]
|
||||||
|
result <- client.expect[String](request).redeemWith(
|
||||||
|
error => IO(Left(error))
|
||||||
|
// .flatTap(_ => IO.println(s"Request failed to url: $url with error: ${error.getMessage}"))
|
||||||
|
,
|
||||||
|
fileContent => IO(Right((fileName, fileContent)))
|
||||||
|
// .flatTap(_ => IO.println(s"Fetched: $fileName"))
|
||||||
|
)
|
||||||
|
} yield result
|
||||||
|
}
|
||||||
|
|
||||||
|
private def fetchFiles(fileNames: List[String]): IO[List[Either[Throwable, (String, String)]]] = {
|
||||||
|
val urls = fileNames.map(baseUrl / _)
|
||||||
|
BlazeClientBuilder[IO](global).resource.use { client =>
|
||||||
|
urls.traverse(url => makeRequest(client, url))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
def fetchInRange(from: LocalDateTime, to: LocalDateTime): IO[List[Either[Throwable, (String, String)]]] = {
|
||||||
|
val fileNames = FileNameService.generate(from, to)
|
||||||
|
fetchFiles(fileNames)
|
||||||
|
}
|
||||||
|
|
||||||
|
def fetchFromDate(date: LocalDate): IO[List[Either[Throwable, (String, String)]]] = {
|
||||||
|
val fileNames = FileNameService.generateFromDate(date)
|
||||||
|
fetchFiles(fileNames)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,24 +1,24 @@
|
|||||||
package fetch
|
package fetch
|
||||||
|
|
||||||
import java.time.format.DateTimeFormatter
|
import java.time.format.DateTimeFormatter
|
||||||
import java.time.{LocalDateTime, Duration}
|
import java.time.{Duration, LocalDate, LocalDateTime}
|
||||||
|
|
||||||
object FileName {
|
object FileNameService {
|
||||||
private val interval = Duration.ofMinutes(30)
|
private val interval = Duration.ofMinutes(30)
|
||||||
private val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
|
private val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
|
||||||
|
|
||||||
def generateLastHour(): List[String] = {
|
|
||||||
// TODO IO.timer() IOTimer
|
|
||||||
val now = LocalDateTime.now
|
|
||||||
generate(now.minusHours(1), now)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO side effect
|
// TODO side effect
|
||||||
def generateLastNHours(hours: Int, now: LocalDateTime = LocalDateTime.now): List[String] = {
|
def generateLastNHours(hours: Int, now: LocalDateTime = LocalDateTime.now): List[String] = {
|
||||||
val roundNow = roundToInterval(now, false)
|
val roundNow = roundToInterval(now, false)
|
||||||
generate(roundNow.minusHours(hours), roundNow)
|
generate(roundNow.minusHours(hours), roundNow)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def generateFromDate(date: LocalDate): List[String] = {
|
||||||
|
val from = LocalDateTime.of(date.getYear, date.getMonth, date.getDayOfMonth, 0, 0)
|
||||||
|
val to = LocalDateTime.of(date.getYear, date.getMonth, date.getDayOfMonth, 23, 59)
|
||||||
|
generate(from, to)
|
||||||
|
}
|
||||||
|
|
||||||
private def roundToInterval(time: LocalDateTime, roundUp: Boolean): LocalDateTime = {
|
private def roundToInterval(time: LocalDateTime, roundUp: Boolean): LocalDateTime = {
|
||||||
val minutes = time.getMinute
|
val minutes = time.getMinute
|
||||||
val adjustment = if (roundUp) (interval.toMinutes - minutes % interval.toMinutes) % interval.toMinutes
|
val adjustment = if (roundUp) (interval.toMinutes - minutes % interval.toMinutes) % interval.toMinutes
|
||||||
@@ -0,0 +1,27 @@
|
|||||||
|
package fetch
|
||||||
|
|
||||||
|
import cats.effect.IO
|
||||||
|
import cats.effect.unsafe.implicits.global
|
||||||
|
import cats.implicits.toTraverseOps
|
||||||
|
import db.DBService
|
||||||
|
|
||||||
|
import java.time.{LocalDate, LocalDateTime}
|
||||||
|
|
||||||
|
object Main {
|
||||||
|
def run: IO[Unit] = {
|
||||||
|
val from = LocalDateTime.of(2023, 4, 23, 20, 0)
|
||||||
|
val to = LocalDateTime.of(2023, 4, 23, 23, 30)
|
||||||
|
for {
|
||||||
|
fetched <- FetchService.fetchInRange(from, to)
|
||||||
|
// fetched <- FetchService.fetchFromDate(LocalDate.of(2023, 4, 23))
|
||||||
|
(errors, successfulDownloads) = fetched.partitionMap(identity)
|
||||||
|
_ <- IO.println(s"Fetch errors: $errors")
|
||||||
|
saveResults <- successfulDownloads.traverse { case (name, content) => DBService.save(name, content) }
|
||||||
|
_ <- IO.println(s"Save results: $saveResults")
|
||||||
|
} yield ()
|
||||||
|
}
|
||||||
|
|
||||||
|
def main(args: Array[String]): Unit = {
|
||||||
|
run.unsafeRunSync()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,14 +1,21 @@
|
|||||||
package server
|
package server
|
||||||
|
|
||||||
import cats.effect._
|
import cats.effect._
|
||||||
|
import cats.implicits.toTraverseOps
|
||||||
|
import db.DBService
|
||||||
|
import fetch.FetchService
|
||||||
|
import io.circe.{Json, Printer}
|
||||||
import org.http4s._
|
import org.http4s._
|
||||||
import org.http4s.dsl.io._
|
import org.http4s.dsl.io._
|
||||||
import org.http4s.implicits._
|
import org.http4s.implicits._
|
||||||
import org.http4s.server.Router
|
import org.http4s.server.Router
|
||||||
import org.http4s.server.blaze.BlazeServerBuilder
|
import org.http4s.server.blaze.BlazeServerBuilder
|
||||||
|
import io.circe.syntax._
|
||||||
|
import io.circe.generic.auto._
|
||||||
|
import org.http4s.circe.jsonEncoder
|
||||||
import parse.{Meteo, Parser}
|
import parse.{Meteo, Parser}
|
||||||
|
|
||||||
import java.time.LocalDateTime
|
import java.time.{LocalDate, LocalDateTime}
|
||||||
import java.time.format.DateTimeFormatter
|
import java.time.format.DateTimeFormatter
|
||||||
import scala.util.Try
|
import scala.util.Try
|
||||||
|
|
||||||
@@ -45,8 +52,34 @@ object Server extends IOApp {
|
|||||||
case _ => BadRequest(s"Invalid request format")
|
case _ => BadRequest(s"Invalid request format")
|
||||||
}
|
}
|
||||||
|
|
||||||
case GET -> Root / "fetch" / dateRange =>
|
// http://localhost:3000/fetchDate/20230423
|
||||||
???
|
case GET -> Root / "fetchDate" / dateStr => {
|
||||||
|
val dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd")
|
||||||
|
val maybeDate = Try(LocalDate.parse(dateStr, dateFormatter)).toEither
|
||||||
|
maybeDate match {
|
||||||
|
case Right(date) => {
|
||||||
|
val result = for {
|
||||||
|
fetched <- FetchService.fetchFromDate(date)
|
||||||
|
(fetchErrors, successDownloads) = fetched.partitionMap(identity)
|
||||||
|
saved <- successDownloads.traverse { case (name, content) => DBService.save(name, content) }
|
||||||
|
(saveErrors, successSaves) = saved.partitionMap(identity)
|
||||||
|
successes = successDownloads.map(s => s"fetched: ${s._1}") ++ successSaves.map(s => s"saved: $s")
|
||||||
|
errors = fetchErrors.map(e => s"FetchError: ${e.getMessage}") ++ saveErrors.map(e => s"SaveError: ${e.getMessage}")
|
||||||
|
} yield (successes, errors)
|
||||||
|
|
||||||
|
result.flatMap { case (successes, errors) =>
|
||||||
|
val responseBody = Json.obj(
|
||||||
|
"errors" -> errors.asJson,
|
||||||
|
"successes" -> successes.asJson
|
||||||
|
)
|
||||||
|
val printer = Printer.spaces2.copy(dropNullValues = true)
|
||||||
|
val prettyJson = printer.print(responseBody)
|
||||||
|
Ok(prettyJson)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
case Left(_) => BadRequest("Invalid request format")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
case GET -> Root / "show" / "fetched_dates" =>
|
case GET -> Root / "show" / "fetched_dates" =>
|
||||||
???
|
???
|
||||||
|
|||||||
Reference in New Issue
Block a user