1
0
mirror of https://github.com/pcvolkmer/etl-processor.git synced 2025-07-06 00:12:53 +00:00

feat: support multiple request content types (#109)

This commit is contained in:
2025-04-06 22:17:46 +02:00
committed by GitHub
parent 8e3de6a220
commit c6b37fda69
15 changed files with 643 additions and 331 deletions

View File

@ -26,7 +26,6 @@ import dev.dnpm.etl.processor.PatientId
import dev.dnpm.etl.processor.services.RequestProcessor
import dev.pcvolkmer.mv64e.mtb.Mtb
import org.slf4j.LoggerFactory
import org.springframework.http.HttpStatus
import org.springframework.http.MediaType
import org.springframework.http.ResponseEntity
import org.springframework.web.bind.annotation.*
@ -47,10 +46,10 @@ class MtbFileRestController(
@PostMapping( consumes = [ MediaType.APPLICATION_JSON_VALUE ] )
fun mtbFile(@RequestBody mtbFile: MtbFile): ResponseEntity<Unit> {
if (mtbFile.consent.status == Consent.Status.ACTIVE) {
logger.debug("Accepted MTB File for processing")
logger.debug("Accepted MTB File (bwHC V1) for processing")
requestProcessor.processMtbFile(mtbFile)
} else {
logger.debug("Accepted MTB File and process deletion")
logger.debug("Accepted MTB File (bwHC V1) and process deletion")
val patientId = PatientId(mtbFile.patient.id)
requestProcessor.processDeletion(patientId)
}
@ -59,7 +58,9 @@ class MtbFileRestController(
@PostMapping( consumes = [ CustomMediaType.APPLICATION_VND_DNPM_V2_MTB_JSON_VALUE] )
fun mtbFile(@RequestBody mtbFile: Mtb): ResponseEntity<Unit> {
return ResponseEntity.status(HttpStatus.NOT_IMPLEMENTED).build()
logger.debug("Accepted MTB File (DNPM V2) for processing")
requestProcessor.processMtbFile(mtbFile)
return ResponseEntity.accepted().build()
}
@DeleteMapping(path = ["{patientId}"])

View File

@ -1,7 +1,7 @@
/*
* This file is part of ETL-Processor
*
* Copyright (c) 2024 Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
* Copyright (c) 2025 Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published
@ -22,10 +22,12 @@ package dev.dnpm.etl.processor.output
import com.fasterxml.jackson.databind.ObjectMapper
import de.ukw.ccc.bwhc.dto.Consent
import de.ukw.ccc.bwhc.dto.MtbFile
import dev.dnpm.etl.processor.RequestId
import dev.dnpm.etl.processor.CustomMediaType
import dev.dnpm.etl.processor.config.KafkaProperties
import dev.dnpm.etl.processor.monitoring.RequestStatus
import org.apache.kafka.clients.producer.ProducerRecord
import org.slf4j.LoggerFactory
import org.springframework.http.MediaType
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.retry.support.RetryTemplate
@ -38,14 +40,20 @@ class KafkaMtbFileSender(
private val logger = LoggerFactory.getLogger(KafkaMtbFileSender::class.java)
override fun send(request: MtbFileSender.MtbFileRequest): MtbFileSender.Response {
override fun <T> send(request: MtbFileRequest<T>): MtbFileSender.Response {
return try {
return retryTemplate.execute<MtbFileSender.Response, Exception> {
val result = kafkaTemplate.send(
kafkaProperties.outputTopic,
key(request),
objectMapper.writeValueAsString(Data(request.requestId, request.mtbFile))
)
val record =
ProducerRecord(kafkaProperties.outputTopic, key(request), objectMapper.writeValueAsString(request))
when (request) {
is BwhcV1MtbFileRequest -> record.headers()
.add("contentType", MediaType.APPLICATION_JSON_VALUE.toByteArray())
is DnpmV2MtbFileRequest -> record.headers()
.add("contentType", CustomMediaType.APPLICATION_VND_DNPM_V2_MTB_JSON_VALUE.toByteArray())
}
val result = kafkaTemplate.send(record)
if (result.get() != null) {
logger.debug("Sent file via KafkaMtbFileSender")
MtbFileSender.Response(RequestStatus.UNKNOWN)
@ -59,7 +67,7 @@ class KafkaMtbFileSender(
}
}
override fun send(request: MtbFileSender.DeleteRequest): MtbFileSender.Response {
override fun send(request: DeleteRequest): MtbFileSender.Response {
val dummyMtbFile = MtbFile.builder()
.withConsent(
Consent.builder()
@ -71,12 +79,15 @@ class KafkaMtbFileSender(
return try {
return retryTemplate.execute<MtbFileSender.Response, Exception> {
val result = kafkaTemplate.send(
kafkaProperties.outputTopic,
key(request),
objectMapper.writeValueAsString(Data(request.requestId, dummyMtbFile))
)
val record =
ProducerRecord(
kafkaProperties.outputTopic,
key(request),
// Always use old BwhcV1FileRequest with Consent REJECT
objectMapper.writeValueAsString(BwhcV1MtbFileRequest(request.requestId, dummyMtbFile))
)
val result = kafkaTemplate.send(record)
if (result.get() != null) {
logger.debug("Sent deletion request via KafkaMtbFileSender")
MtbFileSender.Response(RequestStatus.UNKNOWN)
@ -94,13 +105,12 @@ class KafkaMtbFileSender(
return "${this.kafkaProperties.servers} (${this.kafkaProperties.outputTopic}/${this.kafkaProperties.outputResponseTopic})"
}
private fun key(request: MtbFileSender.MtbFileRequest): String {
return "{\"pid\": \"${request.mtbFile.patient.id}\"}"
private fun key(request: MtbRequest): String {
return when (request) {
is BwhcV1MtbFileRequest -> "{\"pid\": \"${request.content.patient.id}\"}"
is DnpmV2MtbFileRequest -> "{\"pid\": \"${request.content.patient.id}\"}"
is DeleteRequest -> "{\"pid\": \"${request.patientId.value}\"}"
else -> throw IllegalArgumentException("Unsupported request type: ${request::class.simpleName}")
}
}
private fun key(request: MtbFileSender.DeleteRequest): String {
return "{\"pid\": \"${request.patientId.value}\"}"
}
data class Data(val requestId: RequestId, val content: MtbFile)
}
}

View File

@ -19,25 +19,17 @@
package dev.dnpm.etl.processor.output
import de.ukw.ccc.bwhc.dto.MtbFile
import dev.dnpm.etl.processor.PatientPseudonym
import dev.dnpm.etl.processor.RequestId
import dev.dnpm.etl.processor.monitoring.RequestStatus
import org.springframework.http.HttpStatusCode
interface MtbFileSender {
fun send(request: MtbFileRequest): Response
fun <T> send(request: MtbFileRequest<T>): Response
fun send(request: DeleteRequest): Response
fun endpoint(): String
data class Response(val status: RequestStatus, val body: String = "")
data class MtbFileRequest(val requestId: RequestId, val mtbFile: MtbFile)
data class DeleteRequest(val requestId: RequestId, val patientId: PatientPseudonym)
}
fun Int.asRequestStatus(): RequestStatus {
@ -51,4 +43,4 @@ fun Int.asRequestStatus(): RequestStatus {
fun HttpStatusCode.asRequestStatus(): RequestStatus {
return this.value().asRequestStatus()
}
}

View File

@ -0,0 +1,59 @@
/*
* This file is part of ETL-Processor
*
* Copyright (c) 2025 Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published
* by the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package dev.dnpm.etl.processor.output
import de.ukw.ccc.bwhc.dto.MtbFile
import dev.dnpm.etl.processor.PatientPseudonym
import dev.dnpm.etl.processor.RequestId
import dev.pcvolkmer.mv64e.mtb.Mtb
interface MtbRequest {
val requestId: RequestId
}
sealed interface MtbFileRequest<out T> : MtbRequest {
override val requestId: RequestId
val content: T
fun patientPseudonym(): PatientPseudonym
}
data class BwhcV1MtbFileRequest(
override val requestId: RequestId,
override val content: MtbFile
) : MtbFileRequest<MtbFile> {
override fun patientPseudonym(): PatientPseudonym {
return PatientPseudonym(content.patient.id)
}
}
data class DnpmV2MtbFileRequest(
override val requestId: RequestId,
override val content: Mtb
) : MtbFileRequest<Mtb> {
override fun patientPseudonym(): PatientPseudonym {
return PatientPseudonym(content.patient.id)
}
}
data class DeleteRequest(
override val requestId: RequestId,
val patientId: PatientPseudonym
) : MtbRequest

View File

@ -1,7 +1,7 @@
/*
* This file is part of ETL-Processor
*
* Copyright (c) 2024 Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
* Copyright (c) 2025 Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published
@ -19,10 +19,11 @@
package dev.dnpm.etl.processor.output
import dev.dnpm.etl.processor.config.RestTargetProperties
import dev.dnpm.etl.processor.monitoring.RequestStatus
import dev.dnpm.etl.processor.CustomMediaType
import dev.dnpm.etl.processor.PatientPseudonym
import dev.dnpm.etl.processor.config.RestTargetProperties
import dev.dnpm.etl.processor.monitoring.ReportService
import dev.dnpm.etl.processor.monitoring.RequestStatus
import dev.dnpm.etl.processor.monitoring.asRequestStatus
import org.slf4j.LoggerFactory
import org.springframework.http.HttpEntity
@ -46,11 +47,11 @@ abstract class RestMtbFileSender(
abstract fun deleteUrl(patientId: PatientPseudonym): String
override fun send(request: MtbFileSender.MtbFileRequest): MtbFileSender.Response {
override fun <T> send(request: MtbFileRequest<T>): MtbFileSender.Response {
try {
return retryTemplate.execute<MtbFileSender.Response, Exception> {
val headers = getHttpHeaders()
val entityReq = HttpEntity(request.mtbFile, headers)
val headers = getHttpHeaders(request)
val entityReq = HttpEntity(request.content, headers)
val response = restTemplate.postForEntity(
sendUrl(),
entityReq,
@ -76,10 +77,10 @@ abstract class RestMtbFileSender(
return MtbFileSender.Response(RequestStatus.ERROR, "Sonstiger Fehler bei der Übertragung")
}
override fun send(request: MtbFileSender.DeleteRequest): MtbFileSender.Response {
override fun send(request: DeleteRequest): MtbFileSender.Response {
try {
return retryTemplate.execute<MtbFileSender.Response, Exception> {
val headers = getHttpHeaders()
val headers = getHttpHeaders(request)
val entityReq = HttpEntity(null, headers)
restTemplate.delete(
deleteUrl(request.patientId),
@ -102,11 +103,15 @@ abstract class RestMtbFileSender(
return this.restTargetProperties.uri.orEmpty()
}
private fun getHttpHeaders(): HttpHeaders {
private fun getHttpHeaders(request: MtbRequest): HttpHeaders {
val username = restTargetProperties.username
val password = restTargetProperties.password
val headers = HttpHeaders()
headers.contentType = MediaType.APPLICATION_JSON
headers.contentType = when (request) {
is BwhcV1MtbFileRequest -> MediaType.APPLICATION_JSON
is DnpmV2MtbFileRequest -> CustomMediaType.APPLICATION_VND_DNPM_V2_MTB_JSON
else -> MediaType.APPLICATION_JSON
}
if (username.isNullOrBlank() || password.isNullOrBlank()) {
return headers
@ -116,4 +121,4 @@ abstract class RestMtbFileSender(
return headers
}
}
}

View File

@ -1,7 +1,7 @@
/*
* This file is part of ETL-Processor
*
* Copyright (c) 2023 Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
* Copyright (c) 2025 Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published
@ -27,10 +27,11 @@ import dev.dnpm.etl.processor.monitoring.Report
import dev.dnpm.etl.processor.monitoring.Request
import dev.dnpm.etl.processor.monitoring.RequestStatus
import dev.dnpm.etl.processor.monitoring.RequestType
import dev.dnpm.etl.processor.output.MtbFileSender
import dev.dnpm.etl.processor.output.*
import dev.dnpm.etl.processor.pseudonym.PseudonymizeService
import dev.dnpm.etl.processor.pseudonym.anonymizeContentWith
import dev.dnpm.etl.processor.pseudonym.pseudonymizeWith
import dev.pcvolkmer.mv64e.mtb.Mtb
import org.apache.commons.codec.binary.Base32
import org.apache.commons.codec.digest.DigestUtils
import org.springframework.context.ApplicationEventPublisher
@ -55,29 +56,40 @@ class RequestProcessor(
fun processMtbFile(mtbFile: MtbFile, requestId: RequestId) {
val pid = PatientId(mtbFile.patient.id)
mtbFile pseudonymizeWith pseudonymizeService
mtbFile anonymizeContentWith pseudonymizeService
val request = BwhcV1MtbFileRequest(requestId, transformationService.transform(mtbFile))
saveAndSend(request, pid)
}
val request = MtbFileSender.MtbFileRequest(requestId, transformationService.transform(mtbFile))
fun processMtbFile(mtbFile: Mtb) {
processMtbFile(mtbFile, randomRequestId())
}
val patientPseudonym = PatientPseudonym(request.mtbFile.patient.id)
fun processMtbFile(mtbFile: Mtb, requestId: RequestId) {
val pid = PatientId(mtbFile.patient.id)
mtbFile pseudonymizeWith pseudonymizeService
mtbFile anonymizeContentWith pseudonymizeService
val request = DnpmV2MtbFileRequest(requestId, transformationService.transform(mtbFile))
saveAndSend(request, pid)
}
private fun <T> saveAndSend(request: MtbFileRequest<T>, pid: PatientId) {
requestService.save(
Request(
requestId,
patientPseudonym,
request.requestId,
request.patientPseudonym(),
pid,
fingerprint(request.mtbFile),
fingerprint(request),
RequestType.MTB_FILE,
RequestStatus.UNKNOWN
)
)
if (appConfigProperties.duplicationDetection && isDuplication(mtbFile)) {
if (appConfigProperties.duplicationDetection && isDuplication(request)) {
applicationEventPublisher.publishEvent(
ResponseEvent(
requestId,
request.requestId,
Instant.now(),
RequestStatus.DUPLICATION
)
@ -89,7 +101,7 @@ class RequestProcessor(
applicationEventPublisher.publishEvent(
ResponseEvent(
requestId,
request.requestId,
Instant.now(),
responseStatus.status,
when (responseStatus.status) {
@ -100,8 +112,11 @@ class RequestProcessor(
)
}
private fun isDuplication(pseudonymizedMtbFile: MtbFile): Boolean {
val patientPseudonym = PatientPseudonym(pseudonymizedMtbFile.patient.id)
private fun <T> isDuplication(pseudonymizedMtbFileRequest: MtbFileRequest<T>): Boolean {
val patientPseudonym = when (pseudonymizedMtbFileRequest) {
is BwhcV1MtbFileRequest -> PatientPseudonym(pseudonymizedMtbFileRequest.content.patient.id)
is DnpmV2MtbFileRequest -> PatientPseudonym(pseudonymizedMtbFileRequest.content.patient.id)
}
val lastMtbFileRequestForPatient =
requestService.lastMtbFileRequestForPatientPseudonym(patientPseudonym)
@ -109,7 +124,7 @@ class RequestProcessor(
return null != lastMtbFileRequestForPatient
&& !isLastRequestDeletion
&& lastMtbFileRequestForPatient.fingerprint == fingerprint(pseudonymizedMtbFile)
&& lastMtbFileRequestForPatient.fingerprint == fingerprint(pseudonymizedMtbFileRequest)
}
fun processDeletion(patientId: PatientId) {
@ -131,7 +146,7 @@ class RequestProcessor(
)
)
val responseStatus = sender.send(MtbFileSender.DeleteRequest(requestId, patientPseudonym))
val responseStatus = sender.send(DeleteRequest(requestId, patientPseudonym))
applicationEventPublisher.publishEvent(
ResponseEvent(
@ -160,8 +175,11 @@ class RequestProcessor(
}
}
private fun fingerprint(mtbFile: MtbFile): Fingerprint {
return fingerprint(objectMapper.writeValueAsString(mtbFile))
private fun <T> fingerprint(request: MtbFileRequest<T>): Fingerprint {
return when (request) {
is BwhcV1MtbFileRequest -> fingerprint(objectMapper.writeValueAsString(request.content))
is DnpmV2MtbFileRequest -> fingerprint(objectMapper.writeValueAsString(request.content))
}
}
private fun fingerprint(s: String): Fingerprint {
@ -172,4 +190,4 @@ class RequestProcessor(
)
}
}
}

View File

@ -23,10 +23,21 @@ import com.fasterxml.jackson.databind.ObjectMapper
import com.jayway.jsonpath.JsonPath
import com.jayway.jsonpath.PathNotFoundException
import de.ukw.ccc.bwhc.dto.MtbFile
import dev.pcvolkmer.mv64e.mtb.Mtb
class TransformationService(private val objectMapper: ObjectMapper, private val transformations: List<Transformation>) {
fun transform(mtbFile: MtbFile): MtbFile {
var json = objectMapper.writeValueAsString(mtbFile)
val json = transform(objectMapper.writeValueAsString(mtbFile))
return objectMapper.readValue(json, MtbFile::class.java)
}
fun transform(mtbFile: Mtb): Mtb {
val json = transform(objectMapper.writeValueAsString(mtbFile))
return objectMapper.readValue(json, Mtb::class.java)
}
private fun transform(content: String): String {
var json = content
transformations.forEach { transformation ->
val jsonPath = JsonPath.parse(json)
@ -48,7 +59,7 @@ class TransformationService(private val objectMapper: ObjectMapper, private val
json = jsonPath.jsonString()
}
return objectMapper.readValue(json, MtbFile::class.java)
return json
}
fun getTransformations(): List<Transformation> {