131 خطوط
8.0 KiB
Python
131 خطوط
8.0 KiB
Python
"""Two PCB creation endpoints using the existing durable queue and capabilities."""
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import secrets
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Annotated, Literal
|
|
from fastapi import File, Form, Header, UploadFile
|
|
from fastapi.responses import FileResponse, JSONResponse, Response
|
|
from sqlalchemy import select, func, text
|
|
from starlette.concurrency import run_in_threadpool
|
|
from .config import settings
|
|
from .expiry import expiration
|
|
from .db import Job, sessions, active_jobs
|
|
from .storage import storage
|
|
from .uploads import read_pcb
|
|
from .models.pcb import PCBData, PCBStatus, PCBUploadResponse, PCBComparisonStatus, PCBComparisonUploadResponse
|
|
|
|
FRONTEND = Path(__file__).resolve().parents[2] / 'frontend'
|
|
|
|
|
|
def register_pcb_routes(app, *, authorize, require_api, error):
|
|
def state(job, comparison=False):
|
|
base = settings().public_url
|
|
api = f'/api/v1/pcbs/comparisons/{job.id}' if comparison else f'/api/v1/pcbs/{job.id}'
|
|
view = f'/compare/pcb/{job.id}' if comparison else f'/view/pcb/{job.id}'
|
|
result = dict(success=job.status != 'failed', id=job.id, status=job.status,
|
|
document_type='pcb_comparison' if comparison else 'pcb',
|
|
viewer_url=base+view, status_url=base+api+('' if comparison else '/status'),
|
|
access=job.access, expires_at=job.expires_at)
|
|
if comparison: result.update(comparison_id=job.id, url=base+view)
|
|
if job.summary: result.update(json.loads(job.summary))
|
|
if job.error_code: result['error'] = dict(code=job.error_code, message=job.error_message)
|
|
return result
|
|
|
|
async def create(files, access, expires_in, comparison):
|
|
now=time.time();expires_at=expiration(expires_in,now)
|
|
sources = {}
|
|
try:
|
|
for field, file in files.items(): sources[field] = await read_pcb(file)
|
|
finally:
|
|
for file in files.values(): await file.close()
|
|
identifier = secrets.token_hex(24); token = secrets.token_urlsafe(32)
|
|
input_name = 'pcb-comparison.json' if comparison else 'input.PcbDoc'
|
|
job = Job(id=identifier, filename='PCB comparison' if comparison else sources['file'][0],
|
|
status='queued', access=access, token_hash=hashlib.sha256(token.encode()).hexdigest(),
|
|
created_at=now, updated_at=now, expires_at=expires_at, attempt=0, input_key=f'{identifier}/{input_name}')
|
|
if comparison:
|
|
bodies = {f'{identifier}/{side}.PcbDoc':sources[side+'_file'][1] for side in ('old','new')}
|
|
bodies[job.input_key] = json.dumps(dict(schema_version='1.0',document_type='pcb_comparison',
|
|
old_filename=sources['old_file'][0],new_filename=sources['new_file'][0])).encode()
|
|
else: bodies = {job.input_key:sources['file'][1]}
|
|
def persist():
|
|
written = []
|
|
try:
|
|
with sessions().begin() as db:
|
|
if db.bind.dialect.name == 'postgresql': db.execute(text('SELECT pg_advisory_xact_lock(84721309)'))
|
|
pending = db.scalar(select(func.count()).select_from(Job).where(Job.status.in_(['queued','processing']),active_jobs(now)))
|
|
if pending >= settings().max_pending: error('queue_full','Rendering queue is full; try again later',503)
|
|
for key, body in bodies.items():
|
|
# Include the attempted key so a write-then-raise backend is also cleaned up.
|
|
written.append(key)
|
|
try: storage().put(key,body)
|
|
except Exception: error('storage_unavailable','Upload storage unavailable',503)
|
|
db.add(job); db.flush()
|
|
except Exception:
|
|
for key in written:
|
|
try: storage().delete(key)
|
|
except Exception: pass
|
|
raise
|
|
await run_in_threadpool(persist)
|
|
result = state(job,comparison)
|
|
if access == 'private':
|
|
result['viewer_token'] = token; result['viewer_url'] += '#token='+token
|
|
if comparison: result['url'] = result['viewer_url']
|
|
return JSONResponse(result,status_code=202,headers={'Location':result['status_url'],'Retry-After':'1'})
|
|
|
|
@app.post('/api/v1/pcbs/render', status_code=202, response_model=PCBUploadResponse, response_model_exclude_none=True, tags=['PCB'])
|
|
async def render(file:Annotated[UploadFile,File()], access:Annotated[Literal['private','unlisted'],Form()]='private',
|
|
expires_in:Annotated[int|None,Form(ge=-1, description="-1 for no expiry, or positive seconds")]=None, x_api_key:Annotated[str|None,Header()]=None):
|
|
require_api(x_api_key)
|
|
return await create({'file':file},access,expires_in,False)
|
|
|
|
@app.post('/api/v1/pcbs/comparisons',status_code=202,response_model=PCBComparisonUploadResponse,response_model_exclude_none=True,tags=['PCB'])
|
|
async def compare(old_file:Annotated[UploadFile,File()],new_file:Annotated[UploadFile,File()],
|
|
access:Annotated[Literal['private','unlisted'],Form()]='private',
|
|
expires_in:Annotated[int|None,Form(ge=-1, description="-1 for no expiry, or positive seconds")]=None,x_api_key:Annotated[str|None,Header()]=None):
|
|
require_api(x_api_key)
|
|
return await create({'old_file':old_file,'new_file':new_file},access,expires_in,True)
|
|
|
|
def get_job(identifier,key,token,comparison=False):
|
|
return authorize(identifier,key,token,resource_kind='pcb_comparison' if comparison else 'pcb_render')
|
|
|
|
def artifact(job,name):
|
|
if job.status == 'failed': error(job.error_code,job.error_message,422)
|
|
if job.status != 'completed': error('not_ready','PCB processing is still in progress',409)
|
|
try: body = storage().get(job.output_prefix+'/'+name)
|
|
except Exception: error('storage_unavailable','PCB artifact unavailable',503)
|
|
return Response(body,media_type='application/json' if name.endswith('.json') else 'image/svg+xml')
|
|
|
|
@app.get('/api/v1/pcbs/comparisons/{identifier}',response_model=PCBComparisonStatus,response_model_exclude_none=True,tags=['PCB'])
|
|
def comparison_status(identifier:str,x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None):
|
|
job=get_job(identifier,x_api_key,x_viewer_token,True); result=state(job,True)
|
|
if job.status == 'completed':
|
|
try: result.update(json.loads(storage().get(job.output_prefix+'/data.json')))
|
|
except Exception: error('storage_unavailable','PCB comparison unavailable',503)
|
|
return result
|
|
|
|
@app.get('/api/v1/pcbs/comparisons/{identifier}/svg/{side}',response_class=Response,tags=['PCB'])
|
|
def comparison_svg(identifier:str,side:Literal['old','new'],x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None):
|
|
return artifact(get_job(identifier,x_api_key,x_viewer_token,True),side+'.svg')
|
|
|
|
@app.get('/api/v1/pcbs/{identifier}/status',response_model=PCBStatus,response_model_exclude_none=True,tags=['PCB'])
|
|
def status(identifier:str,x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None):
|
|
return state(get_job(identifier,x_api_key,x_viewer_token))
|
|
|
|
@app.get('/api/v1/pcbs/{identifier}/data',response_model=PCBData,tags=['PCB'])
|
|
def data(identifier:str,x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None):
|
|
return artifact(get_job(identifier,x_api_key,x_viewer_token),'data.json')
|
|
|
|
@app.get('/api/v1/pcbs/{identifier}/svg',response_class=Response,tags=['PCB'])
|
|
def svg(identifier:str,x_api_key:Annotated[str|None,Header()]=None,x_viewer_token:Annotated[str|None,Header()]=None):
|
|
return artifact(get_job(identifier,x_api_key,x_viewer_token),'drawing.svg')
|
|
|
|
@app.get('/view/pcb/{identifier}',include_in_schema=False)
|
|
@app.get('/compare/pcb/{identifier}',include_in_schema=False)
|
|
def page(identifier:str):
|
|
if not re.fullmatch('[a-f0-9]{48}',identifier): error('not_found','PCB not found',404)
|
|
return FileResponse(FRONTEND/'pcb.html')
|