Advanced Scala for comprehensions

ForComprehensions.scala

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Await
import scala.concurrent.duration._
import java.util.Calendar

case class Movie(name: String, year: Int, rating: BigDecimal, directorId: Int)

case class Director(id: Int, name: String, yearBorn: Int)

def getMoviesByRating(rating: BigDecimal): Future[Seq[Movie]] = {
  val fakeDbOfMovies = Seq(
    Movie("The Shawshank Redemption", 1994, 9.2, 1),
    Movie("The Godfather", 1972, 9.2, 2),
    Movie("The Godfather: Part II", 1974, 9.0, 2),
    Movie("The Dark Knight", 2008, 9.0, 3),
    Movie("12 Angry Men", 1957, 8.9, 4),
    Movie("Schindler's List", 1993, 8.9, 5),
    Movie("Pulp Fiction", 1994, 8.9, 6),
    Movie("The Lord of the Rings: The Return of the King", 2003, 8.9, 7),
    Movie("The Good, the Bad, and the Ugly", 1966, 8.9, 8),
    Movie("Fight Club", 1999, 8.8, 9)
  )
  Future.successful(fakeDbOfMovies.filter(_.rating == rating))
}

def getDirector(directorId: Int): Future[Option[Director]] = {
  val fakeDbOfDirectors = Seq(
    Director(1, "Frank Darabont", 1959),
    Director(2, "Francis Ford Coppola", 1939),
    Director(3, "Christopher Nolan", 1970),
    Director(4, "Sidney Lumet", 1924),
    Director(5, "Steven Spielberg", 1946),
    Director(6, "Quentin Tarantino", 1963),
    Director(7, "Peter Jackson", 1961),
    Director(8, "Sergio Leone", 1929),
    Director(9, "David Fincher", 1962),
  )
  Future.successful(fakeDbOfDirectors.find(_.id == directorId))
}

def getAverageAgeOfDirectorForRatingNested(rating: BigDecimal): Future[Int] = {
  val currentYear = Calendar.getInstance().get(Calendar.YEAR)

  getMoviesByRating(rating).flatMap(moviesWithRating => {
    val directorAgesForRating = moviesWithRating.map(director => {
      getDirector(director.directorId).map(maybeDirector => {
        maybeDirector.map(director => {
          currentYear - director.yearBorn
        })
      })
    })

    Future.sequence(directorAgesForRating).map(_.flatten[Int].sum / directorAgesForRating.length)
  })
}

def getAverageAgeOfDirectorForRating(rating: BigDecimal): Future[Int] = {
  val currentYear = Calendar.getInstance().get(Calendar.YEAR)

  for {
    moviesWithRating <- getMoviesByRating(rating)
    directorIdsForRating <- Future.successful(moviesWithRating.map(_.directorId))
    maybeDirectors <- Future.sequence(directorIdsForRating.map(getDirector))
    directors <- Future.successful(maybeDirectors.flatten[Director])
    directorBirthYears <- Future.successful(directors.map(_.yearBorn))
    directorAges <- Future.successful(directorBirthYears.map(currentYear - _))
    averageDirectorAge <- Future.successful(directorAges.sum / directors.length)
  } yield averageDirectorAge
}

val averageAge1 = Await.result(getAverageAgeOfDirectorForRatingNested(8.9), atMost = 5.seconds)
val averageAge2 = Await.result(getAverageAgeOfDirectorForRating(8.9), atMost = 5.seconds)
println(averageAge1)
println(averageAge2)