269 lines
9.8 KiB
Python
269 lines
9.8 KiB
Python
"""S3 helpers — boto3 client class, upload/delete, permanent object URLs.
|
|
|
|
No FastAPI imports (house rule). Raise S3ServiceError; s3/views.py maps to HTTPException.
|
|
|
|
Permanent links: we NEVER return expiring presigned URLs. The URL is the virtual-hosted
|
|
HTTPS object address, which stays valid until the object is deleted (or the bucket
|
|
policy stops public GetObject).
|
|
|
|
CV keys are record-scoped (atomicity): DB row is created first, then upload uses that id:
|
|
|
|
Email/{table_record_id}/{user_id}/{file_name}.pdf
|
|
Manual/{table_record_id}/{user_id}/{file_name}.pdf
|
|
Form/{table_record_id}/{recruiter_id}/{file_name}.pdf
|
|
|
|
Callers that create the row MUST delete it if upload_for_record fails.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import mimetypes
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import boto3
|
|
from botocore.client import BaseClient
|
|
from botocore.exceptions import BotoCoreError,ClientError
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
logger=logging.getLogger("s3.plugins")
|
|
|
|
AWS_ACCESS_KEY_ID=os.getenv("AWS_ACCESS_KEY_ID","").strip()
|
|
AWS_SECRET_ACCESS_KEY=os.getenv("AWS_SECRET_ACCESS_KEY","").strip()
|
|
AWS_REGION=(os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "us-east-2").strip()
|
|
S3_BUCKET=os.getenv("S3_BUCKET","").strip()
|
|
# Optional CDN / custom domain. Blank → https://{bucket}.s3.{region}.amazonaws.com/{key}
|
|
S3_PUBLIC_BASE_URL=os.getenv("S3_PUBLIC_BASE_URL","").strip().rstrip("/")
|
|
|
|
# modern buckets often have ACLs disabled; leave blank and rely on bucket policy.
|
|
S3_OBJECT_ACL=os.getenv("S3_OBJECT_ACL","").strip() # e.g. public-read
|
|
|
|
_SAFE_NAME=re.compile(r"[^A-Za-z0-9._-]+")
|
|
_PDF_MIME=frozenset({"application/pdf","application/x-pdf"})
|
|
|
|
|
|
class S3Source:
|
|
"""Top-level folder names — keep spelling exact for console browsing."""
|
|
EMAIL="Email"
|
|
MANUAL="Manual"
|
|
FORM="Form"
|
|
ALL=frozenset({EMAIL,MANUAL,FORM})
|
|
|
|
|
|
class S3ServiceError(Exception):
|
|
"""Raised for config / boto failures — views translate to HTTPException."""
|
|
|
|
def __init__(self,message,status_code=500):
|
|
super().__init__(message)
|
|
self.message=str(message)
|
|
self.status_code=int(status_code)
|
|
|
|
|
|
def sanitize_filename(name: str) -> str:
|
|
raw=(name or "").strip() or "file"
|
|
base=Path(raw).name
|
|
cleaned=_SAFE_NAME.sub("_",base).strip("._") or "file"
|
|
return cleaned[:180]
|
|
|
|
|
|
def guess_content_type(filename: str,fallback: str="application/octet-stream") -> str:
|
|
guessed,_=mimetypes.guess_type(filename or "")
|
|
return guessed or fallback
|
|
|
|
|
|
def assert_pdf(filename: str,content_type: str | None=None) -> str:
|
|
"""Gate: only .pdf (and PDF MIME when provided). Returns sanitized basename."""
|
|
safe=sanitize_filename(filename)
|
|
if not safe.lower().endswith(".pdf"):
|
|
raise S3ServiceError("Only PDF files are allowed",status_code=415)
|
|
mime=(content_type or "").strip().lower().split(";")[0].strip()
|
|
# browsers sometimes send application/octet-stream for PDFs — allow that
|
|
# only when the extension already passed; reject every other non-PDF MIME.
|
|
if mime and mime not in _PDF_MIME and mime!="application/octet-stream":
|
|
raise S3ServiceError(f"Only PDF MIME types are allowed (got {mime})",status_code=415)
|
|
return safe
|
|
|
|
|
|
def normalize_source(source: str) -> str:
|
|
raw=(source or "").strip()
|
|
if not raw:
|
|
raise S3ServiceError("source is required (Email|Manual|Form)",status_code=422)
|
|
# accept case-insensitive input, store canonical folder casing
|
|
for name in S3Source.ALL:
|
|
if raw.lower()==name.lower():
|
|
return name
|
|
raise S3ServiceError(
|
|
f"source must be one of {', '.join(sorted(S3Source.ALL))}",
|
|
status_code=422,
|
|
)
|
|
|
|
|
|
class S3:
|
|
"""One boto3 client + bucket config — upload / delete / URL / health share this."""
|
|
|
|
def __init__(self,client: BaseClient | None=None):
|
|
self._require_config()
|
|
self.bucket=S3_BUCKET
|
|
self.region=AWS_REGION
|
|
self.public_base_url=S3_PUBLIC_BASE_URL
|
|
self.object_acl=S3_OBJECT_ACL
|
|
self.client=client or boto3.client(
|
|
"s3",
|
|
region_name=self.region,
|
|
aws_access_key_id=AWS_ACCESS_KEY_ID,
|
|
aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
|
|
)
|
|
|
|
@staticmethod
|
|
def _require_config():
|
|
missing=[name for name,val in (
|
|
("AWS_ACCESS_KEY_ID",AWS_ACCESS_KEY_ID),
|
|
("AWS_SECRET_ACCESS_KEY",AWS_SECRET_ACCESS_KEY),
|
|
("S3_BUCKET",S3_BUCKET),
|
|
) if not val]
|
|
if missing:
|
|
raise S3ServiceError(
|
|
f"S3 is not configured — set {', '.join(missing)} in backend/.env",
|
|
status_code=500,
|
|
)
|
|
|
|
def _raise_boto(self,exc,action,key=None,status_code=502):
|
|
"""Map ClientError / BotoCoreError → S3ServiceError (single place)."""
|
|
if isinstance(exc,ClientError):
|
|
code=(exc.response or {}).get("Error",{}).get("Code") or ""
|
|
logger.exception("s3 %s failed key=%s code=%s",action,key,code)
|
|
raise S3ServiceError(f"S3 {action} failed: {code or exc}",status_code=status_code) from exc
|
|
logger.exception("s3 %s botocore failure key=%s",action,key)
|
|
raise S3ServiceError(f"S3 {action} failed: {exc}",status_code=status_code) from exc
|
|
|
|
def build_record_object_key(
|
|
self,
|
|
*,
|
|
source: str,
|
|
record_id,
|
|
owner_id,
|
|
filename: str,
|
|
) -> str:
|
|
"""{Email|Manual|Form}/{table_record_id}/{user_or_recruiter_id}/{file}.pdf"""
|
|
folder=normalize_source(source)
|
|
rid=str(record_id or "").strip()
|
|
oid=str(owner_id or "").strip()
|
|
if not rid:
|
|
raise S3ServiceError("table_record_id is required before S3 upload",status_code=422)
|
|
if not oid:
|
|
raise S3ServiceError("owner_id (user_id / recruiter_id) is required before S3 upload",status_code=422)
|
|
safe=assert_pdf(filename)
|
|
return f"{folder}/{rid}/{oid}/{safe}"
|
|
|
|
def permanent_object_url(self,key: str) -> str:
|
|
"""Stable HTTPS URL for a public object — does not expire."""
|
|
object_key=(key or "").lstrip("/")
|
|
if not object_key:
|
|
raise S3ServiceError("object key is required",status_code=422)
|
|
if self.public_base_url:
|
|
return f"{self.public_base_url}/{object_key}"
|
|
if not self.bucket:
|
|
raise S3ServiceError("S3_BUCKET is not configured",status_code=500)
|
|
return f"https://{self.bucket}.s3.{self.region}.amazonaws.com/{object_key}"
|
|
|
|
def upload_bytes(
|
|
self,
|
|
body: bytes,
|
|
filename: str,
|
|
*,
|
|
content_type: str | None=None,
|
|
key: str | None=None,
|
|
) -> dict:
|
|
"""PutObject + permanent URL. Prefer upload_for_record for CV flows."""
|
|
if body is None:
|
|
raise S3ServiceError("file body is required",status_code=422)
|
|
if not key:
|
|
raise S3ServiceError(
|
|
"object key is required — use upload_for_record after the DB row exists",
|
|
status_code=422,
|
|
)
|
|
safe=assert_pdf(filename,content_type)
|
|
object_key=key.lstrip("/")
|
|
ctype=content_type if (content_type or "").strip().lower().startswith("application/pdf") else "application/pdf"
|
|
extra={}
|
|
if self.object_acl:
|
|
extra["ACL"]=self.object_acl
|
|
try:
|
|
self.client.put_object(
|
|
Bucket=self.bucket,
|
|
Key=object_key,
|
|
Body=body,
|
|
ContentType=ctype,
|
|
**extra,
|
|
)
|
|
except (ClientError,BotoCoreError) as e:
|
|
self._raise_boto(e,"upload",key=object_key)
|
|
url=self.permanent_object_url(object_key)
|
|
return {
|
|
"bucket":self.bucket,
|
|
"key":object_key,
|
|
"url":url,
|
|
"content_type":ctype,
|
|
"size":len(body),
|
|
"filename":safe,
|
|
}
|
|
|
|
def upload_for_record(
|
|
self,
|
|
body: bytes,
|
|
filename: str,
|
|
*,
|
|
source: str,
|
|
record_id,
|
|
owner_id,
|
|
content_type: str | None=None,
|
|
) -> dict:
|
|
"""Atomic CV path: requires an existing table row id, then PutObject.
|
|
|
|
Callers MUST roll back (delete) the table row if this raises.
|
|
"""
|
|
key=self.build_record_object_key(
|
|
source=source,
|
|
record_id=record_id,
|
|
owner_id=owner_id,
|
|
filename=filename,
|
|
)
|
|
result=self.upload_bytes(body,filename,content_type=content_type,key=key)
|
|
result["source"]=normalize_source(source)
|
|
result["record_id"]=str(record_id)
|
|
result["owner_id"]=str(owner_id)
|
|
return result
|
|
|
|
def delete_object(self,key: str) -> dict:
|
|
"""DeleteObject — after this the permanent URL 404s."""
|
|
object_key=(key or "").lstrip("/")
|
|
if not object_key:
|
|
raise S3ServiceError("object key is required",status_code=422)
|
|
try:
|
|
self.client.delete_object(Bucket=self.bucket,Key=object_key)
|
|
except (ClientError,BotoCoreError) as e:
|
|
self._raise_boto(e,"delete",key=object_key)
|
|
return {"bucket":self.bucket,"key":object_key,"deleted":True}
|
|
|
|
def head_bucket(self) -> dict:
|
|
"""Reachability probe — credentials + bucket exist."""
|
|
try:
|
|
self.client.head_bucket(Bucket=self.bucket)
|
|
except ClientError as e:
|
|
code=(e.response or {}).get("Error",{}).get("Code") or ""
|
|
status=403 if code in ("403","AccessDenied","AllAccessDisabled") else 502
|
|
self._raise_boto(e,"head_bucket",status_code=status)
|
|
except BotoCoreError as e:
|
|
self._raise_boto(e,"head_bucket")
|
|
base=self.public_base_url or f"https://{self.bucket}.s3.{self.region}.amazonaws.com"
|
|
return {
|
|
"bucket":self.bucket,
|
|
"region":self.region,
|
|
"status":"ok",
|
|
"public_base_url":base,
|
|
}
|