AbsaOSS / AbsaOSS/fa-db

Proposal for refactoring (WIP)

Đang mở
#117 0 bình luận 0 reaction 0 người được giao Xem trên GitHub
enhancement work in progress
Ngôn ngữ chính
Scala
Star
2
Fork
0
Merge trung bình
3 ngày 18 giờ
Pull request đã merge (30 ngày)
3

Mô tả

## Background
```scala
abstract class BaseF[I, R, F[_]: Async](functionNameOverride: Option[String])(implicit val schema: DBSchema) {
def toFragmentsSeq: I => Seq[Fragment]

val functionName: String = {
val fn = functionNameOverride.getOrElse(schema.objectNameFromClassName(getClass))
if (schema.schemaName.isEmpty) {
fn
} else {
s"${schema.schemaName}.$fn"
}
}

protected val alias = "FNC"
def fieldsToSelect: Seq[String] = Seq.empty

protected def selectEntry: String = {
val fieldsSeq = fieldsToSelect
if (fieldsSeq.isEmpty) {
"*"
} else {
val aliasToUse = if (alias.isEmpty) {
""
} else {
s"$alias."
}
fieldsToSelect.map(aliasToUse + _).mkString(",")
}
}

protected final def composeFragments(fragments: Seq[Fragment]): Fragment = {
val args = fragments.toList match {
case head :: tail => tail.foldLeft(head)((acc, frag) => acc ++ fr"," ++ frag)
case Nil => fr""
}
sql"SELECT ${Fragment.const(selectEntry)} FROM ${Fragment.const(functionName)}($args) ${Fragment.const(alias)};"
}

protected def executeQuery[T](input: I, query: Fragment => F[T]): F[T] = {
MonadError[F, Throwable]
.catchNonFatal(composeFragments(toFragmentsSeq(input)))
.flatMap(query)
}
}

trait StatusSupport {
def checkStatus[A](statusWithData: FunctionStatusWithData[A]): Either[StatusException, FunctionStatusWithData[A]]
}

trait StandardStatusSupport extends StatusSupport {
override def checkStatus[A](statusWithData: FunctionStatusWithData[A]): Either[StatusException, FunctionStatusWithData[A]] = {
val functionStatus = statusWithData.functionStatus
functionStatus.statusCode / 10 match {
case 1 => Right(statusWithData)
case 2 => Left(ServerMisconfigurationException(functionStatus))
case 3 => Left(DataConflictException(functionStatus))
case 4 => Left(DataNotFoundException(functionStatus))
case 5 | 6 | 7 | 8 => Left(ErrorInDataException(functionStatus))
case 9 => Left(OtherStatusException(functionStatus))
case _ => Left(StatusOutOfRangeException(functionStatus))
}
}
}

class SingleF[I, R, F[_]: Async](
val toFragmentsSeq: I => Seq[Fragment],
val functionNameOverride: Option[String] = None
)(implicit dbSchema: DBSchema, read: Read[R])
extends BaseF[I, R, F](functionNameOverride) {
def apply(input: I)(implicit transactor: Transactor[F]): F[R] = {
executeQuery(input, fragment => fragment.query[R](read).unique.transact(transactor))
}
}

class SingleFConnectionIO[I, R, F[_]: Async](
val toFragmentsSeq: I => Seq[Fragment],
val functionNameOverride: Option[String] = None
)(implicit dbSchema: DBSchema, read: Read[R])
extends BaseF[I, R, F](functionNameOverride) {
def apply(input: I): F[ConnectionIO[R]] = {
executeQuery(input, fragment => Async[F].delay(fragment.query[R](read).unique))
}
}

abstract class SingleFWithStatusSupport[I, R, F[_]: Async](
val toFragmentsSeq: I => Seq[Fragment],
val functionNameOverride: Option[String] = None
)(implicit dbSchema: DBSchema, read: Read[StatusWithData[R]])
extends BaseF[I, R, F](functionNameOverride)
with StatusSupport {
def apply(input: I)(implicit transactor: Transactor[F]): F[Either[StatusException, FunctionStatusWithData[R]]] = {
executeQuery(input, fragment => fragment.query[StatusWithData[R]](read).unique.transact(transactor))
.map(statusWithData =>
FunctionStatusWithData(FunctionStatus(statusWithData.status, statusWithData.statusText), statusWithData.data)
)
.map(functionStatusWithData => checkStatus(functionStatusWithData))
}
}

class MultiF[I, R, F[_]: Async](
val toFragmentsSeq: I => Seq[Fragment],
val functionNameOverride: Option[String] = None
)(implicit dbSchema: DBSchema, read: Read[R])
extends BaseF[I, R, F](functionNameOverride) {
def apply(input: I)(implicit transactor: Transactor[F]): F[Seq[R]] = {
executeQuery(input, fragment => fragment.query[R](read).to[Seq].transact(transactor))
}
}

class MultiFConnectionIO[I, R, F[_]: Async](
val toFragmentsSeq: I => Seq[Fragment],
val functionNameOverride: Option[String] = None
)(implicit dbSchema: DBSchema, read: Read[R])
extends BaseF[I, R, F](functionNameOverride) {
def apply(input: I): F[ConnectionIO[Seq[R]]] = {
executeQuery(input, fragment => Async[F].delay(fragment.query[R](read).to[Seq]))
}
}

class MultiFStreaming[I, R, F[_]: Async](
val toFragmentsSeq: I => Seq[Fragment],
val functionNameOverride: Option[String] = None
)(implicit dbSchema: DBSchema, read: Read[R])
extends BaseF[I, R, F](functionNameOverride) {
def apply(input: I, chunkSize: Int = 512)(implicit transactor: Transactor[F]): fs2.Stream[F, R] = {
fs2.Stream
.eval(MonadError[F, Throwable].catchNonFatal(composeFragments(toFragmentsSeq(input))))
.flatMap(fragment => fragment.query[R](read).streamWithChunkSize(chunkSize).transact(transactor))
}
}

class OptionF[I, R, F[_]: Async](
val toFragmentsSeq: I => Seq[Fragment],
val functionNameOverride: Option[String] = None
)(implicit dbSchema: DBSchema, read: Read[R])
extends BaseF[I, R, F](functionNameOverride) {
def apply(input: I)(implicit transactor: Transactor[F]): F[Option[R]] = {
executeQuery(input, fragment => fragment.query[R](read).to[Seq].map(_.headOption).transact(transactor))
}
}

class OptionFConnectionIO[I, R, F[_]: Async](
val toFragmentsSeq: I => Seq[Fragment],
val functionNameOverride: Option[String] = None
)(implicit dbSchema: DBSchema, read: Read[R])
extends BaseF[I, R, F](functionNameOverride) {
def apply(input: I): F[ConnectionIO[Option[R]]] = {
executeQuery(input, fragment => Async[F].delay(fragment.query[R](read).to[Seq].map(_.headOption)))
}
}

// Usage
import za.co.absa.fadb.naming.implementations.SnakeCaseNaming.Implicits._
implicit object Runs extends DBSchema

case class Actor(name: String)
class GetActor extends SingleF[Int, Actor, IO]((i: Int) => Seq(fr"$i"))
class GetActorWithStatus extends SingleFWithStatusSupport[Int, Actor, IO]((i: Int) => Seq(fr"$i")) with StandardStatusSupport
class GetActorConnectionIO extends SingleFConnectionIO[Int, Actor, IO]((i: Int) => Seq(fr"$i"))

class Repository(getActorF: GetActor, getActorFWithStatus: GetActorWithStatus, getActorC: GetActorConnectionIO)
(implicit transactor: Transactor[IO]) {
def getActor(input: Int): IO[Actor] = getActorF(input)
def getActorWithStatus(input: Int): IO[Either[StatusException, FunctionStatusWithData[Actor]]] = getActorFWithStatus(input)
def groupedExecution(input: Int): IO[Actor] = {
val connectionIO = for {
actorConnectionIO <- getActorC(input)
// here could be another call
} yield actorConnectionIO

connectionIO.flatMap(connectionIO => connectionIO.transact(transactor))
}
}

// If you need to return Future you can always call unsafeToFuture
class Repository(getActorF: GetActor)(implicit transactor: Transactor[IO]) {
def getActor(input: Int): Future[Actor] = getActorF(input).unsafeToFuture
}
```

Hướng dẫn đóng góp

Chưa lập chỉ mục được hướng dẫn đóng góp cho kho mã nguồn này

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.