6a8eb826a6
Refactor object storage initialization to use a dedicated function and add error handling for uploads, including a fallback to local storage. Replit-Commit-Author: Agent Replit-Commit-Session-Id: cd9a7d26-a4e5-4215-975c-c59f4ed1f06d Replit-Commit-Checkpoint-Type: full_checkpoint Replit-Commit-Event-Id: 127466ff-4d39-4e2e-bc28-80d552851c25 Replit-Commit-Screenshot-Url: https://storage.googleapis.com/screenshot-production-us-central1/d0a1d46d-d203-4308-bc6a-312ac7c0243b/cd9a7d26-a4e5-4215-975c-c59f4ed1f06d/05bPjFc
99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
"""
|
|
Object Storage Service for FTC Team Website
|
|
Handles persistent file uploads using Google Cloud Storage (Replit Object Storage)
|
|
"""
|
|
|
|
from google.cloud import storage
|
|
from google.auth import identity_pool
|
|
import os
|
|
import json
|
|
from uuid import uuid4
|
|
from werkzeug.utils import secure_filename
|
|
|
|
REPLIT_SIDECAR_ENDPOINT = "http://127.0.0.1:1106"
|
|
|
|
|
|
def get_replit_credentials():
|
|
"""Create proper credentials for Replit Object Storage"""
|
|
credentials_config = {
|
|
"audience": "replit",
|
|
"subject_token_type": "access_token",
|
|
"token_url": f"{REPLIT_SIDECAR_ENDPOINT}/token",
|
|
"type": "external_account",
|
|
"credential_source": {
|
|
"url": f"{REPLIT_SIDECAR_ENDPOINT}/credential",
|
|
"format": {
|
|
"type": "json",
|
|
"subject_token_field_name": "access_token",
|
|
},
|
|
},
|
|
"universe_domain": "googleapis.com",
|
|
}
|
|
|
|
credentials = identity_pool.Credentials.from_info(credentials_config)
|
|
return credentials
|
|
|
|
|
|
class ObjectStorageService:
|
|
def __init__(self):
|
|
"""Initialize the object storage client with Replit credentials"""
|
|
credentials = get_replit_credentials()
|
|
self.client = storage.Client(
|
|
credentials=credentials,
|
|
project="",
|
|
)
|
|
|
|
self.bucket_name = os.environ.get('OBJECT_STORAGE_BUCKET')
|
|
if not self.bucket_name:
|
|
raise ValueError(
|
|
"OBJECT_STORAGE_BUCKET environment variable not set. "
|
|
"Please create a bucket in the Object Storage pane and set this variable."
|
|
)
|
|
|
|
self.bucket = self.client.bucket(self.bucket_name)
|
|
|
|
def upload_file(self, file, folder='uploads'):
|
|
"""
|
|
Upload a file to object storage
|
|
|
|
Args:
|
|
file: Werkzeug FileStorage object
|
|
folder: Folder name in the bucket (default: 'uploads')
|
|
|
|
Returns:
|
|
str: Public URL of the uploaded file
|
|
"""
|
|
if not file or not file.filename:
|
|
return None
|
|
|
|
original_filename = secure_filename(file.filename)
|
|
file_extension = os.path.splitext(original_filename)[1]
|
|
unique_filename = f"{uuid4()}{file_extension}"
|
|
object_name = f"{folder}/{unique_filename}"
|
|
|
|
blob = self.bucket.blob(object_name)
|
|
blob.upload_from_file(
|
|
file.stream,
|
|
content_type=file.content_type or 'application/octet-stream'
|
|
)
|
|
|
|
blob.make_public()
|
|
|
|
return blob.public_url
|
|
|
|
def get_blob(self, url):
|
|
"""Get a blob object from a URL"""
|
|
if not url or not url.startswith('https://storage.googleapis.com/'):
|
|
return None
|
|
|
|
parts = url.replace(f'https://storage.googleapis.com/{self.bucket_name}/', '')
|
|
return self.bucket.blob(parts)
|
|
|
|
def delete_file(self, url):
|
|
"""Delete a file from object storage"""
|
|
blob = self.get_blob(url)
|
|
if blob and blob.exists():
|
|
blob.delete()
|
|
return True
|
|
return False
|