Skip to content
Snippets Groups Projects
Select Git revision
  • 050944353a2f6d71857e0fc713bb0f3aeca2050d
  • main default protected
  • dev
  • sidecar-to-clean-uploaded-files
  • supabase
  • serve-django-static
  • FastAPI
  • v0.1.0
8 results

api.py

Blame
  • 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"}