Select Git revision

Remi PLANEL authored
api.py 9.08 KiB
import os
import tarfile
from tempfile import mkstemp
from django.conf import settings
from ninja import Router, File, Form
from ninja.files import UploadedFile
from analysis.models import (
Analysis,
AnalysisHistory,
AnalysisWorkflow,
Gene,
Hmmer,
Protein,
System,
)
from django.contrib.sessions.models import Session
from django.middleware.csrf import get_token
from django.views.decorators.csrf import ensure_csrf_cookie
from django.http import JsonResponse, FileResponse
from analysis.model_schemas import (
AnalysisOutSchema,
AnalysisInSchema,
Error,
GeneOutSchema,
HmmerOutSchema,
ProteinOutSchema,
SystemOutSchema,
)
session_expiry = 2628000
router = Router()
@router.get("csrf")
@ensure_csrf_cookie
def csrf(request):
token = get_token(request)
return JsonResponse({"token": token})
@router.get("init-session")
def init_session(request):
if not request.session.session_key:
request.session.create()
request.session.set_expiry(session_expiry)
return JsonResponse(
{"session": "ok", "expire_date": request.session.get_expiry_date()}
)
@router.post("add", response=AnalysisOutSchema)
def add(
request,
genomefile: UploadedFile = File(...),
):
fd, genome_path = mkstemp(
prefix="defense-finder-", suffix="-sequences-input.fasta", dir="/uploaded-files"
)
with open(genome_path, "wb+") as f:
for chunk in genomefile.chunks():
f.write(chunk)
os.close(fd)
request.session.set_expiry(session_expiry)
if request.session.session_key:
session_key = request.session.session_key
else:
request.session.create()
request.session.set_expiry(session_expiry)
session_key = request.session.session_key
session = Session.objects.get(session_key=session_key)
aw = AnalysisWorkflow.objects.get(galaxy_id=settings.GALAXY_WORKFLOW_ID)
input_files = [
genome_path,
]
params = {}
analysis = aw.invoke(session, input_files, params, f"{genomefile.name}")
return analysis
@router.get("/", response=list[AnalysisOutSchema])
def list(request):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
return Analysis.objects.filter(analysis_history__session=session)
else:
return []
@router.get("/{analysis_id}", response={200: AnalysisOutSchema, 404: Error})
def get_analysis(request, analysis_id: int):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
try:
analysis = Analysis.objects.get(
pk=analysis_id, analysis_history__session=session
)
analysis.synchronize()
analysis.load_genes()
analysis.load_systems()
analysis.load_hmmers()
analysis.load_proteins()
except Analysis.DoesNotExist:
return 404, {
"message": f"No analysis {analysis_id} found in current session"
}
else:
return analysis
else:
return 404, {"message": "No session"}
@router.delete("/{analysis_id}", response={204: None, 404: Error})
def delete_analysis(request, analysis_id: int):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
try:
# history = AnalysisHistory.objects.filter(analysis_id=analysis_id).first()
analysis = Analysis.objects.get(
pk=analysis_id, analysis_history__session=session
)
analysis.delete()
except Analysis.DoesNotExist:
return 404, {
"message": f"No analysis {analysis_id} found in current session"
}
else:
return 204, None
else:
return 404, {"message": "No session"}
@router.put("/{analysis_id}", response={200: AnalysisOutSchema, 404: Error})
def update_analysis(request, analysis_id: int, payload: AnalysisInSchema):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
try:
analysis = Analysis.objects.get(
pk=analysis_id, analysis_history__session=session
)
analysis.name = payload.name
analysis.save()
except Analysis.DoesNotExist:
return 404, {
"message": f"No analysis {analysis_id} found in current session"
}
else:
return analysis
else:
return 404, {"message": "No session"}
@router.get("/{analysis_id}/genes", response={200: GeneOutSchema, 404: Error})
def genes(request, analysis_id: int):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
try:
analysis = Analysis.objects.get(
pk=analysis_id, analysis_history__session=session
)
analysis.load_genes()
try:
gene_list = Gene.objects.get(analysis=analysis)
except Gene.DoesNotExist:
return 404, {
"message": f"No genes for analysis {analysis.name} found in current session"
}
else:
return gene_list
except Analysis.DoesNotExist:
return 404, {
"message": f"No analysis '{analysis.name}' found in current session"
}
else:
return 404, {"message": "No session"}
@router.get("/{analysis_id}/systems", response={200: SystemOutSchema, 404: Error})
def systems(request, analysis_id: int):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
try:
analysis = Analysis.objects.get(
pk=analysis_id, analysis_history__session=session
)
analysis.load_systems()
try:
gene_list = System.objects.get(analysis=analysis)
except System.DoesNotExist:
return 404, {
"message": f"No systems for analysis {analysis.name} found in current session"
}
else:
return gene_list
except Analysis.DoesNotExist:
return 404, {
"message": f"No analysis {analysis.name} found in current session"
}
else:
return 404, {"message": "No session"}
@router.get("/{analysis_id}/hmmers", response={200: HmmerOutSchema, 404: Error})
def hmmers(request, analysis_id: int):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
try:
analysis = Analysis.objects.get(
pk=analysis_id, analysis_history__session=session
)
analysis.load_hmmers()
try:
hmmer = Hmmer.objects.get(analysis=analysis)
except Hmmer.DoesNotExist:
return 404, {
"message": f"No hmmer for analysis {analysis.name} found in current session"
}
else:
return hmmer
except Analysis.DoesNotExist:
return 404, {
"message": f"No analysis {analysis.name} found in current session"
}
else:
return 404, {"message": "No session"}
@router.get("/{analysis_id}/proteins", response={200: ProteinOutSchema, 404: Error})
def proteins(request, analysis_id: int):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
try:
analysis = Analysis.objects.get(
pk=analysis_id, analysis_history__session=session
)
analysis.load_proteins()
try:
gene_list = Protein.objects.get(analysis=analysis)
except Protein.DoesNotExist:
return 404, {
"message": f"No proteins for analysis {analysis.name} found in current session"
}
else:
return gene_list
except Analysis.DoesNotExist:
return 404, {
"message": f"No analysis '{analysis.name}' found in current session"
}
else:
return 404, {"message": "No session"}
@router.get("/{analysis_id}/results-archive")
def results_archive(request, analysis_id: int):
session_key = request.session.session_key
if session_key:
session = Session.objects.get(session_key=session_key)
try:
analysis = Analysis.objects.get(
pk=analysis_id, analysis_history__session=session
)
archive_name = analysis.download_results()
return FileResponse(open(archive_name, "rb"), as_attachment=True)
except Analysis.DoesNotExist:
return 404, {
"message": f"No analysis '{analysis.name}' found in current session"
}
else:
return 404, {"message": "No session"}