Coverage for rest_api/serializers.py: 74%

1from collections import OrderedDict

2import datetime

3from typing import Type

5from django.db.models import Model as DjangoModel

6from django_filters.rest_framework import DjangoFilterBackend

7from rest_framework import serializers

9from . import models

12def find_or_create(

13 data, model: Type[DjangoModel], serializer_class: Type[serializers.Serializer]

14) -> DjangoModel:

15 try:

16 return model.objects.get(**data)

17 except model.DoesNotExist:

18 serializer = serializer_class(data=data)

19 serializer.is_valid(raise_exception=True)

20 return serializer.save()

23class PropertySerializer(serializers.ModelSerializer):

24 class Meta:

25 model = models.Property

26 fields = "__all__"

27 filter_backends = [DjangoFilterBackend]

30class AminoAcidMutationSerializer(serializers.ModelSerializer):

31 class Meta:

32 model = models.AminoAcidMutation

33 fields = "__all__"

36class NucleotideMutationSerializer(serializers.ModelSerializer):

37 class Meta:

38 model = models.NucleotideMutation

39 fields = "__all__"

42class Sample2PropertySerializer(serializers.ModelSerializer):

43 value = serializers.SerializerMethodField(read_only=True)

44 name = serializers.SerializerMethodField(read_only=True)

46 class Meta:

47 model = models.Sample2Property

48 fields = ["name", "value"]

50 def get_value(self, obj: models.Sample2Property):

51 return (

52 obj.value_integer

53 or obj.value_float

54 or obj.value_text

55 or obj.value_varchar

56 or obj.value_blob

57 or obj.value_date

58 or obj.value_zip

59 )

61 def get_name(self, obj: models.Sample2Property):

62 return obj.property.name

65class Sample2PropertyBulkCreateOrUpdateSerializer(serializers.ModelSerializer):

66 sample = serializers.PrimaryKeyRelatedField(queryset=models.Sample.objects.all())

67 property = serializers.PrimaryKeyRelatedField(

68 queryset=models.Property.objects.all()

69 )

70 value_integer = serializers.IntegerField(required=False, allow_null=True)

71 value_float = serializers.FloatField(required=False, allow_null=True)

72 value_text = serializers.CharField(required=False, allow_null=True)

73 value_varchar = serializers.CharField(required=False, allow_null=True)

74 value_blob = serializers.CharField(required=False, allow_null=True)

75 value_date = serializers.DateField(required=False, allow_null=True)

76 value_zip = serializers.CharField(required=False, allow_null=True)

78 class Meta:

79 model = models.Sample2Property

80 fields = [

81 "property",

82 "sample",

83 "value_integer",

84 "value_float",

85 "value_text",

86 "value_varchar",

87 "value_blob",

88 "value_date",

89 "value_zip",

90 ]

91 optional_fields = [

92 "value_integer",

93 "value_float",

94 "value_text",

95 "value_varchar",

96 "value_blob",

97 "value_date",

98 "value_zip",

99 ]

100

101 def validate(self, data: OrderedDict):

102 if not any(attr in data for attr in self.Meta.optional_fields):

103 raise serializers.ValidationError(

104 "At least one of the following fields must be provided: "

105 + ", ".join(self.Meta.optional_fields)

106 )

107 datatype = [key for key in data.keys() if key in self.Meta.optional_fields][0]

108 if "property__name" in data:

109 data["property"] = models.Property.objects.get_or_create(

110 name=data.pop("property__name"), datatype=datatype

111 )[0]

112 return data

113

114 def get_unique_together_validators(self):

115 """Overriding method to disable unique together checks"""

116 return []

117

118

119class SequenceSerializer(serializers.ModelSerializer):

120 class Meta:

121 model = models.Sequence

122 fields = "__all__"

123

124

125class SampleSerializer(serializers.ModelSerializer):

126 name = serializers.CharField(required=True)

127 sequences = serializers.PrimaryKeyRelatedField(

128 queryset=models.Sequence.objects.all(), many=True

129 )

130

131 class Meta:

132 model = models.Sample

133 fields = "__all__"

134 lookup_field = "name"

135

136

137class ReferenceSerializer(serializers.ModelSerializer):

138 class Meta:

139 model = models.Reference

140 fields = "__all__"

141

142

143class AlignmentSerializer(serializers.ModelSerializer):

144 class Meta:

145 model = models.Alignment

146 fields = "__all__"

147

148

149class RepliconSerializer(serializers.ModelSerializer):

150 def create(self, validated_data):

151 validated_data["sequence"] = (

152 validated_data["sequence"].strip().upper().replace("U", "T")

153 )

154 return super().create(validated_data)

155

156 class Meta:

157 model = models.Replicon

158 fields = "__all__"

159

160

161class GeneSerializer(serializers.ModelSerializer):

162 def create(self, validated_data):

163 if "sequence" in validated_data:

164 validated_data["sequence"] = (

165 validated_data["sequence"].strip().upper().replace("U", "T")

166 )

167 return super().create(validated_data)

168

169 class Meta:

170 model = models.Gene

171 fields = "__all__"

172

173

174class CDSSerializer(serializers.ModelSerializer):

175 class Meta:

176 model = models.CDS

177 fields = "__all__"

178

179

180class CDSSegmentSerializer(serializers.ModelSerializer):

181 class Meta:

182 model = models.CDSSegment

183 fields = "__all__"

184

185

186class PeptideSerializer(serializers.ModelSerializer):

187 class Meta:

188 model = models.Peptide

189 fields = "__all__"

190

191

192class PeptideSegmentSerializer(serializers.ModelSerializer):

193 class Meta:

194 model = models.PeptideSegment

195 fields = "__all__"

196

197

198class GeneSegmentSerializer(serializers.ModelSerializer):

199 class Meta:

200 model = models.GeneSegment

201 fields = "__all__"

202

203

204class SampleGenomesSerializer(serializers.ModelSerializer):

205 properties = serializers.SerializerMethodField()

206 genomic_profiles = serializers.SerializerMethodField()

207 proteomic_profiles = serializers.SerializerMethodField()

208 sequences = serializers.PrimaryKeyRelatedField(many=True, read_only=True)

209

210 class Meta:

211 model = models.Sample

212 fields = [

213 "id",

214 "name",

215 "sequences",

216 "datahash",

217 "properties",

218 "genomic_profiles",

219 "proteomic_profiles",

220 ]

221

222 def get_properties(self, obj: models.Sample):

223 """

224 Use prefetched_related properties

225 """

226 custom_properties = []

227

228 # Get custom properties from prefetched relations

229 if hasattr(obj, "properties"):

230 for prop in obj.properties.all():

231 value = (

232 prop.value_integer

233 or prop.value_float

234 or prop.value_text

235 or prop.value_varchar

236 or prop.value_blob

237 or prop.value_date

238 or prop.value_zip

239 )

240 custom_properties.append({"name": prop.property.name, "value": value})

241

242 # Get sample fields

243 important_fields = [

244 "collection_date",

245 "country",

246 "host",

247 "lab",

248 "lineage",

249 "sequencing_tech",

250 "zip_code",

251 "genome_completeness",

252 "init_upload_date",

253 "last_update_date",

254 "data_set",

255 ]

256

257 for field_name in important_fields:

258 if hasattr(obj, field_name):

259 value = getattr(obj, field_name)

260 if value:

261 if isinstance(value, datetime.datetime):

262 value = value.strftime("%Y-%m-%d")

263 elif isinstance(value, datetime.date):

264 value = value.strftime("%Y-%m-%d")

265 custom_properties.append({"name": field_name, "value": value})

266

267 # Get length from sequence (sequence-level property)

268 # NOTE: A sample can have multiple sequences, so we'll include all lenghts? or

269 # unique lengths, right now we include unique lengths

270 if hasattr(obj, "sequences"):

271 lengths = set()

272 for sequence in obj.sequences.all():

273 if sequence.length is not None:

274 lengths.add(sequence.length)

275 if lengths:

276 # If all sequences have the same length, show single value

277 # Otherwise, show comma-separated list

278 length_value = (

279 str(list(lengths)[0])

280 if len(lengths) == 1

281 else ",".join(map(str, sorted(lengths)))

282 )

283 custom_properties.append({"name": "length", "value": length_value})

284

285 return custom_properties

286

287 def get_genomic_profiles(self, obj: models.Sample):

288 showNX = self.context.get("showNX", False)

289 genomic_profiles_dict = {}

290

291 for sequence in obj.sequences.all():

292 for alignment in sequence.alignments.all():

293 replicon = alignment.replicon

294 replicon_acc = replicon.accession

295

296 if replicon_acc not in genomic_profiles_dict:

297 genomic_profiles_dict[replicon_acc] = {}

298

299 for mutation in getattr(alignment, "genomic_profiles", []):

300 if not showNX and ("N" in mutation.alt):

301 continue

302

303 annotations = []

304 for annotation in mutation.annotations.all():

305 annotations.append(str(annotation))

306

307 genomic_profiles_dict[replicon_acc][

308 self.create_NT_format(mutation)

309 ] = annotations

310

311 return OrderedDict(sorted(genomic_profiles_dict.items()))

312

313 def get_proteomic_profiles(self, obj: models.Sample):

314 """

315 Use prefetched proteomic_profiles

316 """

317 showNX = self.context.get("showNX", False)

318 proteomic_profiles = {}

319

320 for sequence in obj.sequences.all():

321 for alignment in sequence.alignments.all():

322 cds_mutations = {}

323

324 for mutation in getattr(alignment, "proteomic_profiles", []):

325 if not showNX and ("X" in mutation.alt):

326 continue

327

328 cds = mutation.cds

329 replicon_acc = cds.gene.replicon.accession

330 key = f"{replicon_acc}: {cds.accession}"

331

332 if key not in cds_mutations:

333 cds_mutations[key] = []

334

335 gene_symbol = cds.gene.symbol

336 label = self.define_proteomic_label(

337 mutation, gene_symbol, mutation.start, mutation.end

338 )

339

340 # AminoAcidMutation has no direct annotation link; annotations

341 # are attached to the parent NucleotideMutation(s).

342 annotations = []

343 for parent in mutation.parent.all():

344 for annotation in parent.annotations.all():

345 annotation_str = str(annotation)

346 if annotation_str not in annotations:

347 annotations.append(annotation_str)

348

349 cds_mutations[key].append(

350 (gene_symbol, mutation.start, label, annotations)

351 )

352

353 for key, labels in cds_mutations.items():

354 if key not in proteomic_profiles:

355 proteomic_profiles[key] = {}

356 for item in sorted(labels, key=lambda x: (x[0], x[1])):

357 proteomic_profiles[key][item[2]] = item[3]

358

359 return OrderedDict(sorted(proteomic_profiles.items()))

360

361 def define_proteomic_label(

362 self,

363 mutation: models.AminoAcidMutation,

364 gene_symbol: str,

365 mutation_start: int,

366 mutation_end: int,

367 ):

368 # SNP and INS

369 if mutation.alt != "":

370 label = f"{gene_symbol}:{mutation.ref}{mutation_end}{mutation.alt}"

371 else: # DEL

372 if mutation_end - mutation_start == 1:

373 label = f"{gene_symbol}:del:" + str(mutation_start + 1)

374 else:

375 label = (

376 f"{gene_symbol}:del:"

377 + str(mutation_start + 1)

378 + "-"

379 + str(mutation_end)

380 )

381 return label

382

383 def create_NT_format(self, mutation: models.NucleotideMutation):

384 label = ""

385 # SNP and INS

386 if mutation.alt != "":

387 label = f"{mutation.ref}{mutation.end}{mutation.alt}"

388 if mutation.is_frameshift:

389 label += "*fs"

390 else: # DEL

391 if mutation.end - mutation.start == 1:

392 label = "del:" + str(mutation.start + 1)

393 else:

394 label = "del:" + str(mutation.start + 1) + "-" + str(mutation.end)

395 if mutation.is_frameshift:

396 label += "*fs"

397

398 return label

399

400

401class SampleGenomesSerializerVCF(serializers.ModelSerializer):

402 genomic_profiles = serializers.SerializerMethodField()

403

404 class Meta:

405 model = models.Sample

406 fields = ["id", "name", "genomic_profiles"]

407

408 def get_genomic_profiles(self, obj: models.Sample):

409 showNX = self.context.get("showNX", False)

410 replicon_dict = {}

411 for sequence in obj.sequences.all():

412 replicon_accession = sequence.alignments.first().replicon.accession

413 if replicon_accession not in replicon_dict:

414 replicon_dict[replicon_accession] = []

415 for alignment in sequence.alignments.all():

416 for mutation in alignment.genomic_profiles:

417 if not showNX and ("N" in mutation.alt):

418 continue

419 variant = {}

420 variant["variant.id"] = mutation.id

421 variant["variant.ref"] = mutation.ref

422 variant["variant.alt"] = mutation.alt

423 variant["variant.start"] = mutation.start

424 variant["variant.end"] = mutation.end

425 replicon_dict[replicon_accession].append(variant)

426 return replicon_dict

427

428

429class SampleGenomesExportStreamSerializer(SampleGenomesSerializer):

430 row = serializers.SerializerMethodField()

431 columns = ["name"]

432

433 def get_row(self, obj: models.Sample):

434 showNX = self.context.get("showNX", False)

435 ctx = self.context

436

437 # Initialize caches once

438 if "custom_prop_cache" not in ctx:

439 ctx["custom_prop_cache"] = {}

440 if "genomic_cache" not in ctx:

441 ctx["genomic_cache"] = {}

442 if "proteomic_cache" not in ctx:

443 ctx["proteomic_cache"] = {}

444

445 # Cache custom properties

446 if obj.id not in ctx["custom_prop_cache"]:

447 props = []

448 if hasattr(obj, "properties"):

449 for prop in obj.properties.all():

450 value = (

451 prop.value_integer

452 or prop.value_float

453 or prop.value_text

454 or prop.value_varchar

455 or prop.value_blob

456 or prop.value_date

457 or prop.value_zip

458 )

459 props.append({"name": prop.property.name, "value": value})

460 ctx["custom_prop_cache"][obj.id] = {p["name"]: p["value"] for p in props}

461

462 custom_properties = ctx["custom_prop_cache"][obj.id]

463

464 # Cache genomic profiles

465 if obj.id not in ctx["genomic_cache"]:

466 ctx["genomic_cache"][obj.id] = self.get_genomic_profiles(obj)

467

468 genomic_dict = ctx["genomic_cache"][obj.id]

469

470 # Cache proteomic profiles

471 if obj.id not in ctx["proteomic_cache"]:

472 ctx["proteomic_cache"][obj.id] = self.get_proteomic_profiles(obj)

473

474 proteomic_dict = ctx["proteomic_cache"][obj.id]

475

476 row = []

477 for column in self.columns:

478 if column.startswith("proteomic_profile"):

479 gene_acc = column.split(": ", 1)[1]

480 row.append(", ".join(proteomic_dict.get(gene_acc, [])))

481 elif column.startswith("genomic_profile"):

482 replicon_acc = column.split(": ", 1)[1]

483 muts = genomic_dict.get(replicon_acc)

484 row.append(", ".join(list(muts.keys())) if muts else "")

485 elif column in custom_properties:

486 row.append(custom_properties[column])

487 else:

488 row.append(getattr(obj, column, ""))

489 return row

490

491 class Meta:

492 model = models.Sample

493 fields = ["row"]

494

495

496class LineagesSerializer(serializers.ModelSerializer):

497 class Meta:

498 model = models.Lineage

499 fields = "__all__"

500

501

502class ProcessingJobSerializer(serializers.ModelSerializer):

503 # Map the abbreviations to their full names for the 'status' field

504 status = serializers.CharField(source="get_status_display")

505 entry_time = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S")

506

507 class Meta:

508 model = models.ProcessingJob

509 fields = ["job_name", "status", "entry_time"]

510

511

512class FileProcessingSerializer(serializers.ModelSerializer):

513 class Meta:

514 model = models.FileProcessing

515 fields = ["file_name"] # Add more fields if needed

516

517

518class ImportLogSerializer(serializers.ModelSerializer):

519 type = serializers.CharField(source="get_type_display")

520 updated = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S")

521

522 class Meta:

523 model = models.ImportLog

524 fields = ["type", "updated", "success"]