"""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')