[Chore] Package Export, Blanks for fields, Styling (#421)

Co-authored-by: Tim Lorsbach <tim@lorsba.ch>
Reviewed-on: enviPath/enviPy#421
This commit is contained in:
2026-07-02 20:12:26 +12:00
parent a092d4a558
commit dba6514013
21 changed files with 647 additions and 450 deletions

View File

@ -6,12 +6,15 @@ import logging
import uuid
from collections import defaultdict
from datetime import datetime
from typing import Any, Dict, List, TYPE_CHECKING
from typing import Any, Dict, List, Optional, TYPE_CHECKING
from django.conf import settings as s
from django.db import transaction
from ninja import Schema
from pydantic import HttpUrl
from epdb.models import (
AdditionalInformation,
Compound,
CompoundStructure,
Edge,
@ -28,7 +31,6 @@ from epdb.models import (
Rule,
RuleBasedRelativeReasoning,
Scenario,
SequentialRule,
Setting,
SimpleAmbitRule,
SimpleRDKitRule,
@ -43,408 +45,235 @@ if TYPE_CHECKING:
from epdb.logic import SPathway
class PackageExporter:
def __init__(
self,
package: Package,
include_models: bool = False,
include_external_identifiers: bool = True,
):
self._raw_package = package
self.include_modes = include_models
self.include_external_identifiers = include_external_identifiers
class LicenseExportSchema(Schema):
cc_string: str
link: HttpUrl
image_link: HttpUrl
def do_export(self):
return PackageExporter._export_package_as_json(
self._raw_package, self.include_modes, self.include_external_identifiers
)
##############
# RefSchemas #
##############
class RefExportSchema(Schema):
uuid: str
url: str
@staticmethod
def _export_package_as_json(
package: Package, include_models: bool = False, include_external_identifiers: bool = True
) -> Dict[str, Any]:
def resolve_uuid(obj):
value = obj.get("uuid") if isinstance(obj, dict) else obj.uuid
return str(value)
class RefCompoundExportSchema(RefExportSchema): ...
class RefCompoundStructureExportSchema(RefExportSchema): ...
class RefReactionExportSchema(RefExportSchema): ...
class RefRuleExportSchema(RefExportSchema): ...
class RefNodeExportSchema(RefExportSchema): ...
class RefEdgeExportSchema(RefExportSchema): ...
class RefPathwayExportSchema(RefExportSchema): ...
class RefScenarioExportSchema(RefExportSchema): ...
############
# Compound #
############
class CompoundExportSchema(RefCompoundExportSchema):
name: str
description: str
aliases: List[str]
default_structure: RefCompoundStructureExportSchema
structures: List["CompoundStructureExportSchema"]
scenarios: List[RefScenarioExportSchema]
class CompoundStructureExportSchema(RefCompoundStructureExportSchema):
name: str
description: str
aliases: List[str]
smiles: str
molfile: Optional[str]
normalized_structure: bool
scenarios: List[RefScenarioExportSchema]
############
# Reaction #
############
class ReactionExportSchema(RefReactionExportSchema):
name: str
description: str
aliases: List[str]
educts: List[RefCompoundStructureExportSchema]
products: List[RefCompoundStructureExportSchema]
rules: List[RefRuleExportSchema]
multi_step: bool
medline_references: List[str] | None
scenarios: List[RefScenarioExportSchema]
#########
# Rules #
#########
class RuleExportSchema(RefRuleExportSchema):
name: str
description: str
aliases: List[str]
smirks: str
reactant_filter_smarts: Optional[str]
product_filter_smarts: Optional[str]
scenarios: List[RefScenarioExportSchema]
class ParallelRuleExportSchema(RefRuleExportSchema):
name: str
description: str
aliases: List[str]
simple_rules: List[RefRuleExportSchema]
scenarios: List[RefScenarioExportSchema]
###########################
# Pathway / Nodes / Edges #
###########################
class NodeExportSchema(RefNodeExportSchema):
name: str
description: str
aliases: List[str]
default_node_label: RefCompoundStructureExportSchema
node_labels: List[RefCompoundStructureExportSchema]
depth: int
stereo_removed: bool
scenarios: List[RefScenarioExportSchema]
class EdgeExportSchema(RefEdgeExportSchema):
name: str
description: str
aliases: List[str]
edge_label: RefReactionExportSchema
start_nodes: List[RefNodeExportSchema]
end_nodes: List[RefNodeExportSchema]
scenarios: List[RefScenarioExportSchema]
class PathwayExportSchema(RefPathwayExportSchema):
name: str
description: str
aliases: List[str]
predicted: bool
nodes: List[NodeExportSchema]
edges: List[EdgeExportSchema]
scenarios: List[RefScenarioExportSchema]
class ScenarioExportSchema(RefScenarioExportSchema):
name: str
description: str
scenario_date: str
scenario_type: str
class AdditionalInformationExportSchema(RefExportSchema):
type: str
data: dict
scenario: RefScenarioExportSchema | None = None
attach_object: RefExportSchema | None = None
@staticmethod
def resolve_attach_object(obj):
if isinstance(obj, dict):
if obj.get("attach_object") is None:
return None
return RefExportSchema.model_validate(obj["attach_object"])
return obj.content_object
###########
# Package #
###########
class PackageExportSchema(Schema):
name: str
description: str
uuid: str
reviewed: bool
license: LicenseExportSchema | None
compounds: List[CompoundExportSchema]
reactions: List[ReactionExportSchema]
simple_rules: List[RuleExportSchema]
composite_rules: List[ParallelRuleExportSchema]
pathways: List[PathwayExportSchema]
scenarios: List[ScenarioExportSchema]
additional_information: List[AdditionalInformationExportSchema]
@staticmethod
def resolve_uuid(obj):
value = obj.get("uuid") if isinstance(obj, dict) else obj.uuid
return str(value)
@staticmethod
def resolve_simple_rules(obj):
if isinstance(obj, dict):
result = []
for r in obj.get("simple_rules", []):
result.append(RuleExportSchema.model_validate(r))
return result
return SimpleAmbitRule.objects.filter(package=obj)
@staticmethod
def resolve_composite_rules(obj):
if isinstance(obj, dict):
result = []
for r in obj.get("composite_rules", []):
result.append(ParallelRuleExportSchema.model_validate(r))
return result
return ParallelRule.objects.filter(package=obj)
@staticmethod
def resolve_additional_information(obj):
if isinstance(obj, dict):
result = []
for ai in obj.get("additional_information", []):
result.append(AdditionalInformationExportSchema.model_validate(ai))
return result
return AdditionalInformation.objects.filter(package=obj)
class PackageExporter:
def __init__(self, package: Package):
self._raw_package = package
def do_export(self):
return PackageExporter._export_package_as_json(self._raw_package)
@staticmethod
def _export_package_as_json(package: Package) -> Dict[str, Any]:
"""
Dumps a Package and all its related objects as JSON.
Args:
package: The Package instance to dump
include_models: Whether to include EPModel objects
include_external_identifiers: Whether to include external identifiers
Returns:
Dict containing the complete package data as JSON-serializable structure
"""
def serialize_base_object(
obj, include_aliases: bool = True, include_scenarios: bool = True
) -> Dict[str, Any]:
"""Serialize common EnviPathModel fields"""
base_dict = {
"uuid": str(obj.uuid),
"name": obj.name,
"description": obj.description,
"url": obj.url,
"kv": obj.kv,
}
data = PackageExportSchema.from_orm(package)
# Add aliases if the object has them
if include_aliases and hasattr(obj, "aliases"):
base_dict["aliases"] = obj.aliases
# Add scenarios if the object has them
if include_scenarios and hasattr(obj, "scenarios"):
base_dict["scenarios"] = [
{"uuid": str(s.uuid), "url": s.url} for s in obj.scenarios.all()
]
return base_dict
def serialize_external_identifiers(obj) -> List[Dict[str, Any]]:
"""Serialize external identifiers for an object"""
if not include_external_identifiers or not hasattr(obj, "external_identifiers"):
return []
identifiers = []
for ext_id in obj.external_identifiers.all():
identifier_dict = {
"uuid": str(ext_id.uuid),
"database": {
"uuid": str(ext_id.database.uuid),
"name": ext_id.database.name,
"base_url": ext_id.database.base_url,
},
"identifier_value": ext_id.identifier_value,
"url": ext_id.url,
"is_primary": ext_id.is_primary,
}
identifiers.append(identifier_dict)
return identifiers
# Start with the package itself
result = serialize_base_object(package, include_aliases=True, include_scenarios=True)
result["reviewed"] = package.reviewed
# # Add license information
# if package.license:
# result['license'] = {
# 'uuid': str(package.license.uuid),
# 'name': package.license.name,
# 'link': package.license.link,
# 'image_link': package.license.image_link
# }
# else:
# result['license'] = None
# Initialize collections
result.update(
{
"compounds": [],
"structures": [],
"rules": {"simple_rules": [], "parallel_rules": [], "sequential_rules": []},
"reactions": [],
"pathways": [],
"nodes": [],
"edges": [],
"scenarios": [],
"models": [],
}
)
print(f"Exporting package: {package.name}")
# Export compounds
print("Exporting compounds...")
for compound in package.compounds.prefetch_related("default_structure").order_by("url"):
compound_dict = serialize_base_object(
compound, include_aliases=True, include_scenarios=True
)
if compound.default_structure:
compound_dict["default_structure"] = {
"uuid": str(compound.default_structure.uuid),
"url": compound.default_structure.url,
}
else:
compound_dict["default_structure"] = None
compound_dict["external_identifiers"] = serialize_external_identifiers(compound)
result["compounds"].append(compound_dict)
# Export compound structures
print("Exporting compound structures...")
compound_structures = (
CompoundStructure.objects.filter(compound__package=package)
.select_related("compound")
.order_by("url")
)
for structure in compound_structures:
structure_dict = serialize_base_object(
structure, include_aliases=True, include_scenarios=True
)
structure_dict.update(
{
"compound": {
"uuid": str(structure.compound.uuid),
"url": structure.compound.url,
},
"smiles": structure.smiles,
"canonical_smiles": structure.canonical_smiles,
"inchikey": structure.inchikey,
"normalized_structure": structure.normalized_structure,
"external_identifiers": serialize_external_identifiers(structure),
}
)
result["structures"].append(structure_dict)
# Export rules
print("Exporting rules...")
# Simple rules (including SimpleAmbitRule and SimpleRDKitRule)
for rule in SimpleRule.objects.filter(package=package).order_by("url"):
rule_dict = serialize_base_object(rule, include_aliases=True, include_scenarios=True)
# Add specific fields for SimpleAmbitRule
if isinstance(rule, SimpleAmbitRule):
rule_dict.update(
{
"rule_type": "SimpleAmbitRule",
"smirks": rule.smirks,
"reactant_filter_smarts": rule.reactant_filter_smarts or "",
"product_filter_smarts": rule.product_filter_smarts or "",
}
)
elif isinstance(rule, SimpleRDKitRule):
rule_dict.update(
{"rule_type": "SimpleRDKitRule", "reaction_smarts": rule.reaction_smarts}
)
else:
rule_dict["rule_type"] = "SimpleRule"
result["rules"]["simple_rules"].append(rule_dict)
# Parallel rules
for rule in (
ParallelRule.objects.filter(package=package)
.prefetch_related("simple_rules")
.order_by("url")
):
rule_dict = serialize_base_object(rule, include_aliases=True, include_scenarios=True)
rule_dict["rule_type"] = "ParallelRule"
rule_dict["simple_rules"] = [
{"uuid": str(sr.uuid), "url": sr.url} for sr in rule.simple_rules.all()
]
result["rules"]["parallel_rules"].append(rule_dict)
# Sequential rules
for rule in (
SequentialRule.objects.filter(package=package)
.prefetch_related("simple_rules")
.order_by("url")
):
rule_dict = serialize_base_object(rule, include_aliases=True, include_scenarios=True)
rule_dict["rule_type"] = "SequentialRule"
rule_dict["simple_rules"] = [
{
"uuid": str(sr.uuid),
"url": sr.url,
"order_index": sr.sequentialruleordering_set.get(
sequential_rule=rule
).order_index,
}
for sr in rule.simple_rules.all()
]
result["rules"]["sequential_rules"].append(rule_dict)
# Export reactions
print("Exporting reactions...")
for reaction in package.reactions.prefetch_related("educts", "products", "rules").order_by(
"url"
):
reaction_dict = serialize_base_object(
reaction, include_aliases=True, include_scenarios=True
)
reaction_dict.update(
{
"educts": [{"uuid": str(e.uuid), "url": e.url} for e in reaction.educts.all()],
"products": [
{"uuid": str(p.uuid), "url": p.url} for p in reaction.products.all()
],
"rules": [{"uuid": str(r.uuid), "url": r.url} for r in reaction.rules.all()],
"multi_step": reaction.multi_step,
"medline_references": reaction.medline_references,
"external_identifiers": serialize_external_identifiers(reaction),
}
)
result["reactions"].append(reaction_dict)
# Export pathways
print("Exporting pathways...")
for pathway in package.pathways.order_by("url"):
pathway_dict = serialize_base_object(
pathway, include_aliases=True, include_scenarios=True
)
# Add setting reference if exists
if hasattr(pathway, "setting") and pathway.setting:
pathway_dict["setting"] = {
"uuid": str(pathway.setting.uuid),
"url": pathway.setting.url,
}
else:
pathway_dict["setting"] = None
result["pathways"].append(pathway_dict)
# Export nodes
print("Exporting nodes...")
pathway_nodes = (
Node.objects.filter(pathway__package=package)
.select_related("pathway", "default_node_label")
.prefetch_related("node_labels", "out_edges")
.order_by("url")
)
for node in pathway_nodes:
node_dict = serialize_base_object(node, include_aliases=True, include_scenarios=True)
node_dict.update(
{
"pathway": {"uuid": str(node.pathway.uuid), "url": node.pathway.url},
"default_node_label": {
"uuid": str(node.default_node_label.uuid),
"url": node.default_node_label.url,
},
"node_labels": [
{"uuid": str(label.uuid), "url": label.url}
for label in node.node_labels.all()
],
"out_edges": [
{"uuid": str(edge.uuid), "url": edge.url} for edge in node.out_edges.all()
],
"depth": node.depth,
}
)
result["nodes"].append(node_dict)
# Export edges
print("Exporting edges...")
pathway_edges = (
Edge.objects.filter(pathway__package=package)
.select_related("pathway", "edge_label")
.prefetch_related("start_nodes", "end_nodes")
.order_by("url")
)
for edge in pathway_edges:
edge_dict = serialize_base_object(edge, include_aliases=True, include_scenarios=True)
edge_dict.update(
{
"pathway": {"uuid": str(edge.pathway.uuid), "url": edge.pathway.url},
"edge_label": {"uuid": str(edge.edge_label.uuid), "url": edge.edge_label.url},
"start_nodes": [
{"uuid": str(node.uuid), "url": node.url} for node in edge.start_nodes.all()
],
"end_nodes": [
{"uuid": str(node.uuid), "url": node.url} for node in edge.end_nodes.all()
],
}
)
result["edges"].append(edge_dict)
# Export scenarios
print("Exporting scenarios...")
for scenario in package.scenarios.order_by("url"):
scenario_dict = serialize_base_object(
scenario, include_aliases=False, include_scenarios=False
)
scenario_dict.update(
{
"scenario_date": scenario.scenario_date,
"scenario_type": scenario.scenario_type,
"parent": {"uuid": str(scenario.parent.uuid), "url": scenario.parent.url}
if scenario.parent
else None,
"additional_information": scenario.additional_information,
}
)
result["scenarios"].append(scenario_dict)
# Export models
if include_models:
print("Exporting models...")
package_models = (
package.models.select_related("app_domain")
.prefetch_related("rule_packages", "data_packages", "eval_packages")
.order_by("url")
)
for model in package_models:
model_dict = serialize_base_object(
model, include_aliases=True, include_scenarios=False
)
# Common fields for PackageBasedModel
if hasattr(model, "rule_packages"):
model_dict.update(
{
"rule_packages": [
{"uuid": str(p.uuid), "url": p.url}
for p in model.rule_packages.all()
],
"data_packages": [
{"uuid": str(p.uuid), "url": p.url}
for p in model.data_packages.all()
],
"eval_packages": [
{"uuid": str(p.uuid), "url": p.url}
for p in model.eval_packages.all()
],
"threshold": model.threshold,
"eval_results": model.eval_results,
"model_status": model.model_status,
}
)
if model.app_domain:
model_dict["app_domain"] = {
"uuid": str(model.app_domain.uuid),
"url": model.app_domain.url,
}
else:
model_dict["app_domain"] = None
# Specific fields for different model types
if isinstance(model, RuleBasedRelativeReasoning):
model_dict.update(
{
"model_type": "RuleBasedRelativeReasoning",
"min_count": model.min_count,
"max_count": model.max_count,
}
)
elif isinstance(model, MLRelativeReasoning):
model_dict["model_type"] = "MLRelativeReasoning"
elif isinstance(model, EnviFormer):
model_dict["model_type"] = "EnviFormer"
else:
model_dict["model_type"] = "EPModel"
result["models"].append(model_dict)
print(f"Export completed for package: {package.name}")
print(f"- Compounds: {len(result['compounds'])}")
print(f"- Structures: {len(result['structures'])}")
print(f"- Simple rules: {len(result['rules']['simple_rules'])}")
print(f"- Parallel rules: {len(result['rules']['parallel_rules'])}")
print(f"- Sequential rules: {len(result['rules']['sequential_rules'])}")
print(f"- Reactions: {len(result['reactions'])}")
print(f"- Pathways: {len(result['pathways'])}")
print(f"- Nodes: {len(result['nodes'])}")
print(f"- Edges: {len(result['edges'])}")
print(f"- Scenarios: {len(result['scenarios'])}")
print(f"- Models: {len(result['models'])}")
return result
return data.model_dump(mode="json")
class PackageImporter: