forked from enviPath/enviPy
[Feature] Legacy API Layer (#80)
Co-authored-by: Tim Lorsbach <tim@lorsba.ch> Reviewed-on: enviPath/enviPy#80
This commit is contained in:
739
epdb/legacy_api.py
Normal file
739
epdb/legacy_api.py
Normal file
@ -0,0 +1,739 @@
|
||||
from typing import List, Dict, Optional, Any
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.http import HttpResponse
|
||||
from ninja import Router, Schema, Field, Form
|
||||
|
||||
from utilities.chem import FormatConverter
|
||||
from .logic import PackageManager
|
||||
from .models import Compound, CompoundStructure, Package, User, UserPackagePermission, Rule, Reaction, Scenario, Pathway
|
||||
|
||||
|
||||
def _anonymous_or_real(request):
|
||||
if request.user.is_authenticated and not request.user.is_anonymous:
|
||||
return request.user
|
||||
return get_user_model().objects.get(username='anonymous')
|
||||
|
||||
|
||||
# router = Router(auth=SessionAuth())
|
||||
router = Router()
|
||||
|
||||
|
||||
class Error(Schema):
|
||||
message: str
|
||||
|
||||
|
||||
class SimpleObject(Schema):
|
||||
id: str = Field(None, alias="url")
|
||||
name: str = Field(None, alias="name")
|
||||
reviewStatus: bool = Field(None, alias="package.reviewed")
|
||||
|
||||
|
||||
################
|
||||
# Login/Logout #
|
||||
################
|
||||
class SimpleUser(Schema):
|
||||
id: str = Field(None, alias="url")
|
||||
identifier: str = 'user'
|
||||
name: str = Field(None, alias='username')
|
||||
email: str = Field(None, alias='email')
|
||||
|
||||
|
||||
@router.post("/", response={200: SimpleUser, 403: Error})
|
||||
def login(request, loginusername: Form[str], loginpassword: Form[str], hiddenMethod: Form[str]):
|
||||
from django.contrib.auth import authenticate
|
||||
from django.contrib.auth import login
|
||||
email = User.objects.get(username=loginusername).email
|
||||
user = authenticate(username=email, password=loginpassword)
|
||||
if user:
|
||||
login(request, user)
|
||||
return user
|
||||
else:
|
||||
return 403, {'message': 'Invalid username or password'}
|
||||
|
||||
|
||||
class SimpleGroup(Schema):
|
||||
id: str
|
||||
identifier: str = 'group'
|
||||
name: str
|
||||
|
||||
|
||||
###########
|
||||
# Package #
|
||||
###########
|
||||
class SimplePackage(SimpleObject):
|
||||
identifier: str = 'package'
|
||||
reviewStatus: bool = Field(None, alias="reviewed")
|
||||
|
||||
|
||||
class PackageSchema(Schema):
|
||||
description: str = Field(None, alias="description")
|
||||
id: str = Field(None, alias="url")
|
||||
links: List[Dict[str, List[str | int]]] = Field([], alias="links")
|
||||
name: str = Field(None, alias="name")
|
||||
primaryGroup: Optional[SimpleGroup] = None
|
||||
readers: List[Dict[str, str]] = Field([], alias="readers")
|
||||
reviewComment: str = Field(None, alias="review_comment")
|
||||
reviewStatus: str = Field(None, alias="review_status")
|
||||
writers: List[Dict[str, str]] = Field([], alias="writers")
|
||||
|
||||
@staticmethod
|
||||
def resolve_links(obj: Package):
|
||||
return [
|
||||
{
|
||||
'Pathways': [
|
||||
f'{obj.url}/pathway', obj.pathways.count()
|
||||
]
|
||||
}, {
|
||||
'Rules': [
|
||||
f'{obj.url}/rule', obj.rules.count()
|
||||
]
|
||||
}, {
|
||||
'Compounds': [
|
||||
f'{obj.url}/compound', obj.compounds.count()
|
||||
]
|
||||
}, {
|
||||
'Reactions': [
|
||||
f'{obj.url}/reaction', obj.reactions.count()
|
||||
]
|
||||
}, {
|
||||
'Relative Reasoning': [
|
||||
f'{obj.url}/relative-reasoning', obj.models.count()
|
||||
]
|
||||
}, {
|
||||
'Scenarios': [
|
||||
f'{obj.url}/scenario', obj.scenarios.count()
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def resolve_readers(obj: Package):
|
||||
users = User.objects.filter(
|
||||
id__in=UserPackagePermission.objects.filter(
|
||||
package=obj,
|
||||
permission=UserPackagePermission.READ[0]
|
||||
).values_list('user', flat=True)
|
||||
).distinct()
|
||||
|
||||
return [{u.id: u.name} for u in users]
|
||||
|
||||
@staticmethod
|
||||
def resolve_writers(obj: Package):
|
||||
users = User.objects.filter(
|
||||
id__in=UserPackagePermission.objects.filter(
|
||||
package=obj,
|
||||
permission=UserPackagePermission.WRITE[0]
|
||||
).values_list('user', flat=True)
|
||||
).distinct()
|
||||
|
||||
return [{u.id: u.name} for u in users]
|
||||
|
||||
@staticmethod
|
||||
def resolve_review_comment(obj):
|
||||
return ""
|
||||
|
||||
@staticmethod
|
||||
def resolve_review_status(obj):
|
||||
return 'reviewed' if obj.reviewed else 'unreviewed'
|
||||
|
||||
|
||||
class PackageWrapper(Schema):
|
||||
package: List['PackageSchema']
|
||||
|
||||
|
||||
@router.get("/package", response={200: PackageWrapper, 403: Error})
|
||||
def get_packages(request):
|
||||
return {'package': PackageManager.get_all_readable_packages(request.user, include_reviewed=True)}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}", response={200: PackageSchema, 403: Error})
|
||||
def get_package(request, package_uuid):
|
||||
try:
|
||||
return PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Package with id {package_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
################################
|
||||
# Compound / CompoundStructure #
|
||||
################################
|
||||
class SimpleCompound(SimpleObject):
|
||||
identifier: str = 'compound'
|
||||
|
||||
|
||||
class CompoundPathwayScenario(Schema):
|
||||
scenarioId: str
|
||||
scenarioName: str
|
||||
scenarioType: str
|
||||
|
||||
|
||||
class CompoundSchema(Schema):
|
||||
aliases: List[str] = Field([], alias="aliases")
|
||||
description: str = Field(None, alias="description")
|
||||
externalReferences: Dict[str, List[str]] = Field(None, alias="external_references")
|
||||
id: str = Field(None, alias="url")
|
||||
halflifes: List[Dict[str, str]] = Field([], alias="halflifes")
|
||||
identifier: str = 'compound'
|
||||
imageSize: int = 600
|
||||
name: str = Field(None, alias="name")
|
||||
pathwayScenarios: List[CompoundPathwayScenario] = Field([], alias="pathway_scenarios")
|
||||
pathways: List['SimplePathway'] = Field([], alias="related_pathways")
|
||||
pubchemCompoundReferences: List[str] = Field([], alias="pubchem_compound_references")
|
||||
reactions: List['SimpleReaction'] = Field([], alias="related_reactions")
|
||||
reviewStatus: str = Field(False, alias="review_status")
|
||||
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
|
||||
structures: List['CompoundStructureSchema'] = []
|
||||
|
||||
@staticmethod
|
||||
def resolve_review_status(obj: CompoundStructure):
|
||||
return 'reviewed' if obj.package.reviewed else 'unreviewed'
|
||||
|
||||
@staticmethod
|
||||
def resolve_external_references(obj: Compound):
|
||||
# TODO
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def resolve_structures(obj: Compound):
|
||||
return CompoundStructure.objects.filter(compound=obj)
|
||||
|
||||
@staticmethod
|
||||
def resolve_halflifes(obj: Compound):
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def resolve_pubchem_compound_references(obj: Compound):
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def resolve_pathway_scenarios(obj: Compound):
|
||||
return [
|
||||
{
|
||||
'scenarioId': 'https://envipath.org/package/5882df9c-dae1-4d80-a40e-db4724271456/scenario/cd8350cd-4249-4111-ba9f-4e2209338501',
|
||||
'scenarioName': 'Fritz, R. & Brauner, A. (1989) - (00004)',
|
||||
'scenarioType': 'Soil'
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
class CompoundWrapper(Schema):
|
||||
compound: List['SimpleCompound']
|
||||
|
||||
|
||||
class SimpleCompoundStructure(SimpleObject):
|
||||
identifier: str = 'structure'
|
||||
reviewStatus: bool = Field(None, alias="compound.package.reviewed")
|
||||
|
||||
|
||||
class CompoundStructureSchema(Schema):
|
||||
InChI: str = Field(None, alias="inchi")
|
||||
aliases: List[str] = Field([], alias="aliases")
|
||||
canonicalSmiles: str = Field(None, alias="canonical_smiles")
|
||||
charge: int = Field(None, alias="charge")
|
||||
description: str = Field(None, alias="description")
|
||||
externalReferences: Dict[str, List[str]] = Field(None, alias="external_references")
|
||||
formula: str = Field(None, alias="formula")
|
||||
halflifes: List[Dict[str, str]] = Field([], alias="halflifes")
|
||||
id: str = Field(None, alias="url")
|
||||
identifier: str = 'structure'
|
||||
imageSize: int = 600
|
||||
inchikey: str = Field(None, alias="inchikey")
|
||||
isDefaultStructure: bool = Field(None, alias="is_default_structure")
|
||||
mass: float = Field(None, alias="mass")
|
||||
name: str = Field(None, alias="name")
|
||||
pathways: List['SimplePathway'] = Field([], alias="related_pathways")
|
||||
pubchemCompoundReferences: List[str] = Field([], alias="pubchem_compound_references")
|
||||
reactions: List['SimpleReaction'] = Field([], alias="related_reactions")
|
||||
reviewStatus: str = Field(None, alias="review_status")
|
||||
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
|
||||
smiles: str = Field(None, alias="smiles")
|
||||
|
||||
@staticmethod
|
||||
def resolve_review_status(obj: CompoundStructure):
|
||||
return 'reviewed' if obj.compound.package.reviewed else 'unreviewed'
|
||||
|
||||
@staticmethod
|
||||
def resolve_inchi(obj: CompoundStructure):
|
||||
return FormatConverter.InChI(obj.smiles)
|
||||
|
||||
@staticmethod
|
||||
def resolve_charge(obj: CompoundStructure):
|
||||
print(obj.smiles)
|
||||
print(FormatConverter.charge(obj.smiles))
|
||||
return FormatConverter.charge(obj.smiles)
|
||||
|
||||
@staticmethod
|
||||
def resolve_formula(obj: CompoundStructure):
|
||||
return FormatConverter.formula(obj.smiles)
|
||||
|
||||
@staticmethod
|
||||
def resolve_mass(obj: CompoundStructure):
|
||||
return FormatConverter.mass(obj.smiles)
|
||||
|
||||
@staticmethod
|
||||
def resolve_external_references(obj: CompoundStructure):
|
||||
# TODO
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def resolve_halflifes(obj: CompoundStructure):
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def resolve_pubchem_compound_references(obj: CompoundStructure):
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def resolve_pathway_scenarios(obj: CompoundStructure):
|
||||
return [
|
||||
{
|
||||
'scenarioId': 'https://envipath.org/package/5882df9c-dae1-4d80-a40e-db4724271456/scenario/cd8350cd-4249-4111-ba9f-4e2209338501',
|
||||
'scenarioName': 'Fritz, R. & Brauner, A. (1989) - (00004)',
|
||||
'scenarioType': 'Soil'
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
class CompoundStructureWrapper(Schema):
|
||||
structure: List['SimpleCompoundStructure']
|
||||
|
||||
|
||||
@router.get("/compound", response={200: CompoundWrapper, 403: Error})
|
||||
def get_compounds(request):
|
||||
qs = Compound.objects.none()
|
||||
for p in PackageManager.get_reviewed_packages():
|
||||
qs |= Compound.objects.filter(package=p)
|
||||
return {'compound': qs}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/compound", response={200: CompoundWrapper, 403: Error})
|
||||
def get_package_compounds(request, package_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return {'compound': Compound.objects.filter(package=p).prefetch_related('package')}
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Compounds for Package with id {package_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/compound/{uuid:compound_uuid}", response={200: CompoundSchema, 403: Error})
|
||||
def get_package_compound(request, package_uuid, compound_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return Compound.objects.get(package=p, uuid=compound_uuid)
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Compound with id {compound_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/compound/{uuid:compound_uuid}/structure",
|
||||
response={200: CompoundStructureWrapper, 403: Error})
|
||||
def get_package_compound_structures(request, package_uuid, compound_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return {'structure': Compound.objects.get(package=p, uuid=compound_uuid).structures.all()}
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting CompoundStructures for Compound with id {compound_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/compound/{uuid:compound_uuid}/structure/{uuid:structure_uuid}",
|
||||
response={200: CompoundStructureSchema, 403: Error})
|
||||
def get_package_compound_structure(request, package_uuid, compound_uuid, structure_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return CompoundStructure.objects.get(uuid=structure_uuid,
|
||||
compound=Compound.objects.get(package=p, uuid=compound_uuid))
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting CompoundStructure with id {structure_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
#########
|
||||
# Rules #
|
||||
#########
|
||||
class SimpleRule(SimpleObject):
|
||||
identifier: str = 'rule'
|
||||
|
||||
@staticmethod
|
||||
def resolve_url(obj: Rule):
|
||||
return obj.url.replace('-ambit-', '-').replace('-rdkit-', '-')
|
||||
|
||||
|
||||
class RuleWrapper(Schema):
|
||||
rule: List['SimpleRule']
|
||||
|
||||
|
||||
class SimpleRuleSchema(Schema):
|
||||
aliases: List[str] = Field([], alias="aliases")
|
||||
description: str = Field(None, alias="description")
|
||||
ecNumbers: List[Dict[str, str]] = Field([], alias="ec_numbers")
|
||||
engine: str = 'ambit'
|
||||
id: str = Field(None, alias="url")
|
||||
identifier: str = Field(None, alias="identifier")
|
||||
isCompositeRule: bool = False
|
||||
name: str = Field(None, alias="name")
|
||||
pathways: List['SimplePathway'] = Field([], alias="related_pathways")
|
||||
productFilterSmarts: str = Field("", alias="product_filter_smarts")
|
||||
productSmarts: str = Field(None, alias="products_smarts")
|
||||
reactantFilterSmarts: str = Field("", alias="reactant_filter_smarts")
|
||||
reactantSmarts: str = Field(None, alias="reactants_smarts")
|
||||
reactions: List['SimpleReaction'] = Field([], alias="related_reactions")
|
||||
reviewStatus: str = Field(None, alias="review_status")
|
||||
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
|
||||
smirks: str = Field("", alias="smirks")
|
||||
# TODO
|
||||
transformations: str = Field("", alias="transformations")
|
||||
|
||||
@staticmethod
|
||||
def resolve_url(obj: Rule):
|
||||
return obj.url.replace('-ambit-', '-').replace('-rdkit-', '-')
|
||||
|
||||
@staticmethod
|
||||
def resolve_identifier(obj: Rule):
|
||||
if 'simple-rule' in obj.url:
|
||||
return 'simple-rule'
|
||||
if 'simple-ambit-rule' in obj.url:
|
||||
return 'simple-rule'
|
||||
elif 'parallel-rule' in obj.url:
|
||||
return 'parallel-rule'
|
||||
elif 'sequential-rule' in obj.url:
|
||||
return 'sequential-rule'
|
||||
else:
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def resolve_review_status(obj: Rule):
|
||||
return 'reviewed' if obj.package.reviewed else 'unreviewed'
|
||||
|
||||
@staticmethod
|
||||
def resolve_product_filter_smarts(obj: Rule):
|
||||
return obj.product_filter_smarts if obj.product_filter_smarts else ''
|
||||
|
||||
@staticmethod
|
||||
def resolve_reactant_filter_smarts(obj: Rule):
|
||||
return obj.reactant_filter_smarts if obj.reactant_filter_smarts else ''
|
||||
|
||||
|
||||
class CompositeRuleSchema(Schema):
|
||||
aliases: List[str] = Field([], alias="aliases")
|
||||
description: str = Field(None, alias="description")
|
||||
ecNumbers: List[Dict[str, str]] = Field([], alias="ec_numbers")
|
||||
id: str = Field(None, alias="url")
|
||||
identifier: str = Field(None, alias="identifier")
|
||||
isCompositeRule: bool = True
|
||||
name: str = Field(None, alias="name")
|
||||
pathways: List['SimplePathway'] = Field([], alias="related_pathways")
|
||||
productFilterSmarts: str = Field("", alias="product_filter_smarts")
|
||||
reactantFilterSmarts: str = Field("", alias="reactant_filter_smarts")
|
||||
reactions: List['SimpleReaction'] = Field([], alias="related_reactions")
|
||||
reviewStatus: str = Field(None, alias="review_status")
|
||||
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
|
||||
simpleRules: List['SimpleRule'] = Field([], alias="simple_rules")
|
||||
|
||||
@staticmethod
|
||||
def resolve_ec_numbers(obj: Rule):
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def resolve_url(obj: Rule):
|
||||
return obj.url.replace('-ambit-', '-').replace('-rdkit-', '-')
|
||||
|
||||
@staticmethod
|
||||
def resolve_identifier(obj: Rule):
|
||||
if 'simple-rule' in obj.url:
|
||||
return 'simple-rule'
|
||||
if 'simple-ambit-rule' in obj.url:
|
||||
return 'simple-rule'
|
||||
elif 'parallel-rule' in obj.url:
|
||||
return 'parallel-rule'
|
||||
elif 'sequential-rule' in obj.url:
|
||||
return 'sequential-rule'
|
||||
else:
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def resolve_review_status(obj: Rule):
|
||||
return 'reviewed' if obj.package.reviewed else 'unreviewed'
|
||||
|
||||
@staticmethod
|
||||
def resolve_product_filter_smarts(obj: Rule):
|
||||
return obj.product_filter_smarts if obj.product_filter_smarts else ''
|
||||
|
||||
@staticmethod
|
||||
def resolve_reactant_filter_smarts(obj: Rule):
|
||||
return obj.reactant_filter_smarts if obj.reactant_filter_smarts else ''
|
||||
|
||||
|
||||
@router.get("/rule", response={200: RuleWrapper, 403: Error})
|
||||
def get_rules(request):
|
||||
qs = Rule.objects.none()
|
||||
for p in PackageManager.get_reviewed_packages():
|
||||
qs |= Rule.objects.filter(package=p)
|
||||
return {'rule': qs}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/rule", response={200: RuleWrapper, 403: Error})
|
||||
def get_package_rules(request, package_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return {'rule': Rule.objects.filter(package=p).prefetch_related('package')}
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Rules for Package with id {package_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/rule/{uuid:rule_uuid}",
|
||||
response={200: SimpleRuleSchema | CompositeRuleSchema, 403: Error})
|
||||
def get_package_rule(request, package_uuid, rule_uuid):
|
||||
return _get_package_rule(request, package_uuid, rule_uuid)
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/simple-rule/{uuid:rule_uuid}",
|
||||
response={200: SimpleRuleSchema | CompositeRuleSchema, 403: Error})
|
||||
def get_package_simple_rule(request, package_uuid, rule_uuid):
|
||||
return _get_package_rule(request, package_uuid, rule_uuid)
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/parallel-rule/{uuid:rule_uuid}",
|
||||
response={200: SimpleRuleSchema | CompositeRuleSchema, 403: Error})
|
||||
def get_package_parallel_rule(request, package_uuid, rule_uuid):
|
||||
return _get_package_rule(request, package_uuid, rule_uuid)
|
||||
|
||||
|
||||
def _get_package_rule(request, package_uuid, rule_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return Rule.objects.get(package=p, uuid=rule_uuid)
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Rule with id {rule_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
# POST
|
||||
@router.post("/package/{uuid:package_uuid}/rule/{uuid:rule_uuid}", response={200: str | Any, 403: Error})
|
||||
def post_package_rule(request, package_uuid, rule_uuid, compound: Form[str] = None):
|
||||
return _post_package_rule(request, package_uuid, rule_uuid, compound=compound)
|
||||
|
||||
|
||||
@router.post("/package/{uuid:package_uuid}/simple-rule/{uuid:rule_uuid}", response={200: str | Any, 403: Error})
|
||||
def post_package_simple_rule(request, package_uuid, rule_uuid, compound: Form[str] = None):
|
||||
return _post_package_rule(request, package_uuid, rule_uuid, compound=compound)
|
||||
|
||||
|
||||
@router.post("/package/{uuid:package_uuid}/parallel-rule/{uuid:rule_uuid}", response={200: str | Any, 403: Error})
|
||||
def post_package_parallel_rule(request, package_uuid, rule_uuid, compound: Form[str] = None):
|
||||
return _post_package_rule(request, package_uuid, rule_uuid, compound=compound)
|
||||
|
||||
|
||||
def _post_package_rule(request, package_uuid, rule_uuid, compound: Form[str]):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
r = Rule.objects.get(package=p, uuid=rule_uuid)
|
||||
|
||||
if compound is not None:
|
||||
if not compound.split():
|
||||
return 400, {'message': 'Compound is empty'}
|
||||
|
||||
product_sets = r.apply(compound)
|
||||
|
||||
res = []
|
||||
for p_set in product_sets:
|
||||
for product in p_set:
|
||||
res.append(product)
|
||||
|
||||
return HttpResponse('\n'.join(res), content_type="text/plain")
|
||||
|
||||
return r
|
||||
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Rule with id {rule_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
############
|
||||
# Reaction #
|
||||
############
|
||||
class SimpleReaction(SimpleObject):
|
||||
identifier: str = 'reaction'
|
||||
|
||||
|
||||
class ReactionWrapper(Schema):
|
||||
reaction: List['SimpleReaction']
|
||||
|
||||
|
||||
class ReactionCompoundStructure(Schema):
|
||||
compoundName: str = Field(None, alias="name")
|
||||
id: str = Field(None, alias="url")
|
||||
smiles: str = Field(None, alias="smiles")
|
||||
|
||||
|
||||
class ReactionSchema(Schema):
|
||||
aliases: List[str] = Field([], alias="aliases")
|
||||
description: str = Field(None, alias="description")
|
||||
ecNumbers: List[Dict[str, str]] = Field([], alias="ec_numbers")
|
||||
educts: List['ReactionCompoundStructure'] = Field([], alias="educts")
|
||||
id: str = Field(None, alias="url")
|
||||
identifier: str = 'reaction'
|
||||
medlineRefs: List[str] = Field([], alias="medline_references")
|
||||
multistep: bool = Field(None, alias="multi_step")
|
||||
name: str = Field(None, alias="name")
|
||||
pathways: List['SimplePathway'] = Field([], alias="related_pathways")
|
||||
products: List['ReactionCompoundStructure'] = Field([], alias="products")
|
||||
references: List[Dict[str, List[str]]] = Field([], alias="references")
|
||||
reviewStatus: str = Field(None, alias="review_status")
|
||||
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
|
||||
smirks: str = Field("", alias="smirks")
|
||||
|
||||
@staticmethod
|
||||
def resolve_smirks(obj: Reaction):
|
||||
return obj.smirks()
|
||||
|
||||
@staticmethod
|
||||
def resolve_ec_numbers(obj: Reaction):
|
||||
# TODO fetch via scenario EnzymeAI
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def resolve_references(obj: Reaction):
|
||||
# TODO
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def resolve_medline_references(obj: Reaction):
|
||||
# TODO
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def resolve_review_status(obj: Rule):
|
||||
return 'reviewed' if obj.package.reviewed else 'unreviewed'
|
||||
|
||||
|
||||
@router.get("/reaction", response={200: ReactionWrapper, 403: Error})
|
||||
def get_reactions(request):
|
||||
qs = Reaction.objects.none()
|
||||
for p in PackageManager.get_reviewed_packages():
|
||||
qs |= Reaction.objects.filter(package=p)
|
||||
return {'reaction': qs}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/reaction", response={200: ReactionWrapper, 403: Error})
|
||||
def get_package_reactions(request, package_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return {'reaction': Reaction.objects.filter(package=p).prefetch_related('package')}
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Reactions for Package with id {package_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/reaction/{uuid:reaction_uuid}", response={200: ReactionSchema, 403: Error})
|
||||
def get_package_reaction(request, package_uuid, reaction_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return Reaction.objects.get(package=p, uuid=reaction_uuid)
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Reaction with id {reaction_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
############
|
||||
# Scenario #
|
||||
############
|
||||
class SimpleScenario(SimpleObject):
|
||||
identifier: str = 'scenario'
|
||||
|
||||
|
||||
class ScenarioWrapper(Schema):
|
||||
scenario: List['SimpleScenario']
|
||||
|
||||
|
||||
class ScenarioSchema(Schema):
|
||||
aliases: List[str] = Field([], alias="aliases")
|
||||
collection: Dict['str', List[Dict[str, Any]]] = Field([], alias="collection")
|
||||
collectionID: Optional[str] = None
|
||||
description: str = Field(None, alias="description")
|
||||
id: str = Field(None, alias="url")
|
||||
identifier: str = 'scenario'
|
||||
linkedTo: List[Dict[str, str]] = Field({}, alias="linked_to")
|
||||
name: str = Field(None, alias="name")
|
||||
pathways: List['SimplePathway'] = Field([], alias="related_pathways")
|
||||
relatedScenarios: List[Dict[str, str]] = Field([], alias="related_scenarios")
|
||||
reviewStatus: str = Field(None, alias="review_status")
|
||||
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
|
||||
type: str = Field(None, alias="scenario_type")
|
||||
|
||||
@staticmethod
|
||||
def resolve_collection(obj: Scenario):
|
||||
return obj.additional_information
|
||||
|
||||
@staticmethod
|
||||
def resolve_review_status(obj: Rule):
|
||||
return 'reviewed' if obj.package.reviewed else 'unreviewed'
|
||||
|
||||
|
||||
@router.get("/scenario", response={200: ScenarioWrapper, 403: Error})
|
||||
def get_scenarios(request):
|
||||
qs = Scenario.objects.none()
|
||||
for p in PackageManager.get_reviewed_packages():
|
||||
qs |= Scenario.objects.filter(package=p)
|
||||
return {'scenario': qs}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/scenario", response={200: ScenarioWrapper, 403: Error})
|
||||
def get_package_scenarios(request, package_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return {'scenario': Scenario.objects.filter(package=p).prefetch_related('package')}
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Scenarios for Package with id {package_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/scenario/{uuid:scenario_uuid}", response={200: ScenarioSchema, 403: Error})
|
||||
def get_package_scenario(request, package_uuid, scenario_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return Scenario.objects.get(package=p, uuid=scenario_uuid)
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Scenario with id {scenario_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
###########
|
||||
# Pathway #
|
||||
###########
|
||||
class SimplePathway(SimpleObject):
|
||||
identifier: str = 'pathway'
|
||||
|
||||
class PathwayWrapper(Schema):
|
||||
pathway: List['SimplePathway']
|
||||
|
||||
@router.get("/pathway", response={200: PathwayWrapper, 403: Error})
|
||||
def get_pathways(request):
|
||||
qs = Pathway.objects.none()
|
||||
for p in PackageManager.get_reviewed_packages():
|
||||
qs |= Pathway.objects.filter(package=p)
|
||||
return {'pathway': qs}
|
||||
|
||||
|
||||
@router.get("/package/{uuid:package_uuid}/pathway", response={200: PathwayWrapper, 403: Error})
|
||||
def get_package_pathways(request, package_uuid):
|
||||
try:
|
||||
p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
return {'pathway': Pathway.objects.filter(package=p).prefetch_related('package')}
|
||||
except ValueError:
|
||||
return 403, {
|
||||
'message': f'Getting Pathways for Package with id {package_uuid} failed due to insufficient rights!'}
|
||||
|
||||
|
||||
# @router.get("/package/{uuid:package_uuid}/pathway/{uuid:pathway_uuid}", response={200: Pathway, 403: Error})
|
||||
# def get_package_pathway(request, package_uuid, pathway_uuid):
|
||||
# try:
|
||||
# p = PackageManager.get_package_by_id(request.user, package_uuid)
|
||||
# return Pathway.objects.get(package=p, uuid=pathway_uuid)
|
||||
# except ValueError:
|
||||
# return 403, {
|
||||
# 'message': f'Getting Pathway with id {pathway_uuid} failed due to insufficient rights!'}
|
||||
@ -2,7 +2,6 @@ from django.conf import settings
|
||||
from django.shortcuts import redirect
|
||||
from django.urls import reverse
|
||||
|
||||
|
||||
class LoginRequiredMiddleware:
|
||||
def __init__(self, get_response):
|
||||
self.get_response = get_response
|
||||
@ -11,6 +10,7 @@ class LoginRequiredMiddleware:
|
||||
reverse('logout'),
|
||||
reverse('admin:login'),
|
||||
reverse('admin:index'),
|
||||
'/api/legacy/'
|
||||
] + getattr(settings, 'LOGIN_EXEMPT_URLS', [])
|
||||
|
||||
def __call__(self, request):
|
||||
|
||||
@ -889,6 +889,23 @@ class CompoundStructure(EnviPathModel, AliasMixin, ScenarioMixin, ChemicalIdenti
|
||||
def as_svg(self, width: int = 800, height: int = 400):
|
||||
return IndigoUtils.mol_to_svg(self.smiles, width=width, height=height)
|
||||
|
||||
@property
|
||||
def related_pathways(self):
|
||||
pathways = Node.objects.filter(node_labels__in=[self]).values_list('pathway', flat=True)
|
||||
return Pathway.objects.filter(package=self.compound.package, id__in=set(pathways)).order_by('name')
|
||||
|
||||
@property
|
||||
def related_reactions(self):
|
||||
return (
|
||||
Reaction.objects.filter(package=self.compound.package, educts__in=[self])
|
||||
|
|
||||
Reaction.objects.filter(package=self.compound.package, products__in=[self])
|
||||
).order_by('name')
|
||||
|
||||
@property
|
||||
def is_default_structure(self):
|
||||
return self.compound.default_structure == self
|
||||
|
||||
|
||||
class Rule(PolymorphicModel, EnviPathModel, AliasMixin, ScenarioMixin):
|
||||
package = models.ForeignKey('epdb.Package', verbose_name='Package', on_delete=models.CASCADE, db_index=True)
|
||||
|
||||
Reference in New Issue
Block a user