Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions packages/opencloning/src/opencloning/batch_cloning/index.html
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,8 @@
<body>
<h1>Batch Cloning Methods</h1>
<div class="method">
<a href="/batch_cloning/pombe">S. pombe Gene Cloning</a>
<p>Batch cloning of S. pombe genes using homologous recombination.</p>
<a href="/batch_cloning/yeast_primer_design">Yeast PCR-based Cloning primer design</a>
<p>Design PCR-based cloning strategies for yeast in batch (Bähler et al. 1998).</p>
</div>
<div class="method">
<a href="/batch_cloning/ziqiang_et_al2024">CRISPR Guide RNA Cloning</a>
Expand Down
170 changes: 127 additions & 43 deletions packages/opencloning/src/opencloning/batch_cloning/pombe/__init__.py
Original file line number Diff line number Diff line change
@@ -1,79 +1,163 @@
from fastapi import Form, File, UploadFile, HTTPException
from typing import Annotated, Literal
from tempfile import TemporaryDirectory
import os
from fastapi.responses import FileResponse
from .pombe_get_primers import main as pombe_primers
import tempfile
from fastapi.responses import FileResponse, HTMLResponse
from .pombe_clone import main as pombe_clone
from .pombe_summary import main as pombe_summary
from .pombe_gather import main as pombe_gather
from .pombe_primer_table import build_primer_summary_df, primer_summary_to_html
import shutil
import traceback
from ...get_router import get_router
from fastapi import Request
from opencloning.dna_functions import get_sequence_from_euroscarf_url, request_from_addgene, request_from_snapgene
from pydna.primer import Primer
from pydna.opencloning_models import UploadedFileSource
from pydna.parsers import parse as pydna_parse
from pydna.parsers import parse_snapgene

router = get_router()


@router.get('/batch_cloning/pombe')
@router.get('/batch_cloning/yeast_primer_design')
async def get_batch_cloning_page(request: Request):
return FileResponse(os.path.join(os.path.dirname(__file__), 'index.html'))


@router.post('/batch_cloning/pombe')
DEFAULT_PLASMID_OPTIONS = {
'gene_deletion': {
'kanmx6': ('addgene', '39296', None),
'natmx6': ('snapgene', 'yeast_plasmids', 'pFA6a-natMX6'),
'hphmx6': ('snapgene', 'yeast_plasmids', 'pFA6a-hphMX6'),
},
'gene_cterm_tagging': {
'kanmx6': ('addgene', '87023', None),
'natmx6': ('addgene', '52693', None),
'hphmx6': ('addgene', '105156', None),
},
'promoter_not_tag': {
'kanmx6': ('addgene', '39280', None), # pFA6a-kanMX6-P3nmt1
'natmx6': ('euroscarf', 'P30425', None), # pFA6a-natMX6-P3nmt1
'hphmx6': ('addgene', '105162', None), # pFA6a-hphMX6-3nmt1
},
}


def raise_plasmid_import_error(exception: Exception, mode) -> None:
raise HTTPException(status_code=503, detail=f'Failed to import plasmid from {mode}: {exception}') from exception


@router.post('/batch_cloning/yeast_primer_design')
async def post_batch_cloning(
cloning_type: Annotated[
Literal['gene_deletion', 'gene_cterm_tagging', 'promoter_not_tag', 'promoter_with_tag'], Form(...)
],
desired_output: Annotated[Literal['simulate_cloning', 'primers_only'], Form(...)],
assembly_accession: str = Form(..., pattern=r'^GC[AF]_[0-9.]+$', min_length=1),
gene_list: str = Form(...),
integration_binding_forward: str = Form(..., pattern=r'^[ACGTacgt]+$', min_length=1),
integration_binding_reverse: str = Form(..., pattern=r'^[ACGTacgt]+$', min_length=1),
plasmid_file: UploadFile | None = File(None),
addgene_id: str | None = Form(None),
plasmid_option: Annotated[Literal['addgene', 'file'], Form(...)] = None,
checking_primer_forward: str = Form(..., pattern=r'^[ACGTacgt]+$', min_length=1),
checking_primer_reverse: str = Form(..., pattern=r'^[ACGTacgt]+$', min_length=1),
plasmid_option: Annotated[Literal['addgene', 'file', 'default'], Form(...)] = None,
checking_primer_forward: str = Form('', pattern=r'^[ACGTacgt]*$'),
checking_primer_reverse: str = Form('', pattern=r'^[ACGTacgt]*$'),
resistance_marker: Annotated[Literal['kanmx6', 'natmx6', 'hphmx6', 'other'], Form(...)] = None,
):

plasmid = plasmid_file if plasmid_option == 'file' else addgene_id
if plasmid is None:
raise HTTPException(status_code=400, detail='No plasmid provided')

genes = [gene.strip() for gene in gene_list.split() if gene.strip()]

if not genes:
raise HTTPException(status_code=400, detail='No valid genes provided')

with TemporaryDirectory() as temp_dir:
if plasmid_option == 'file':
# Write the plasmid to the temp dir
with open(os.path.join(temp_dir, plasmid_file.filename), 'wb') as f:
shutil.copyfileobj(plasmid_file.file, f)

# Write the checking primers
with open(os.path.join(temp_dir, 'checking_primers.fa'), 'w') as f:
f.write(f'>common_insert_fwd\n{checking_primer_forward}\n>common_insert_rvs\n{checking_primer_reverse}')

for gene in genes:
try:
await pombe_primers(gene, temp_dir)
except Exception:
raise HTTPException(status_code=404, detail=f'Primers for {gene} not found')
try:
if plasmid_option == 'file':
with open(os.path.join(temp_dir, plasmid_file.filename), 'rb') as f:
await pombe_clone(
gene, 'GCF_000002945.2', temp_dir, UploadFile(file=f, filename=plasmid_file.filename)
)
else:
await pombe_clone(gene, 'GCF_000002945.2', temp_dir, addgene_id)
except Exception:
# Show the stack trace in console
print(f'Error occurred while cloning {gene}:')
traceback.print_exc()
raise HTTPException(status_code=400, detail=f'Clone for {gene} failed')
if desired_output == 'primers_only':
gene_primers = []
try:
for gene in genes:
primers = await pombe_clone(
gene,
assembly_accession,
integration_binding_forward,
integration_binding_reverse,
cloning_type,
primers_only=True,
)
gene_primers.append((gene, cloning_type, primers))
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))

primer_df = build_primer_summary_df(gene_primers)
return HTMLResponse(content=primer_summary_to_html(primer_df))

common_primer_forward = (
Primer(checking_primer_forward, name='common_insert_fwd') if checking_primer_forward else None
)
common_primer_reverse = (
Primer(checking_primer_reverse, name='common_insert_rvs') if checking_primer_reverse else None
)

if plasmid_option == 'default':
try:
mode, first, second = DEFAULT_PLASMID_OPTIONS[cloning_type][resistance_marker]
if mode == 'addgene':
plasmid = await request_from_addgene(first)
elif mode == 'euroscarf':
plasmid = await get_sequence_from_euroscarf_url(first)
else:
plasmid = await request_from_snapgene(first, second)
except KeyError:
raise HTTPException(
status_code=400, detail=f'Resistance marker {resistance_marker} is not supported for default plasmid'
)
except Exception as e:
raise_plasmid_import_error(e, mode)

elif plasmid_option == 'file':
try:
assert plasmid_file is not None
assert plasmid_file.filename is not None
file_content = await plasmid_file.read()
if plasmid_file.filename.endswith('.dna'):
plasmid = parse_snapgene(file_content)[0]
else:
plasmid = pydna_parse(file_content.decode('utf-8'))[0]
plasmid.source = UploadedFileSource(
file_name=plasmid_file.filename,
sequence_file_format=plasmid.annotations['pydna_parse_sequence_file_format'],
index_in_file=0,
)
except Exception as e:
raise HTTPException(status_code=400, detail=f'Plasmid loading failed: {e}')

if plasmid_option == 'addgene':
assert addgene_id is not None
try:
plasmid = await request_from_addgene(addgene_id)
except Exception as e:
raise_plasmid_import_error(e, 'addgene')

with tempfile.TemporaryDirectory() as temp_dir:
try:
for gene in genes:
await pombe_clone(
gene,
assembly_accession,
integration_binding_forward,
integration_binding_reverse,
cloning_type,
output_dir=temp_dir,
plasmid=plasmid,
common_primer_forward=common_primer_forward,
common_primer_reverse=common_primer_reverse,
)
except ValueError as e:
raise HTTPException(status_code=400, detail=f'Cloning failed: {e}')

try:
pombe_summary(temp_dir)
pombe_gather(temp_dir)
except Exception as e:
raise HTTPException(status_code=400, detail=f'Summary failed: {e}')

# zip the temp dir and return it
zip_filename = f'{temp_dir}_archive'
shutil.make_archive(zip_filename, 'zip', temp_dir)
zip_file = f'{zip_filename}.zip'
Expand Down
Loading
Loading