I have been facing lots of issue when i decided to switch to uploading my file in frappe to linode bucket.
The file is getting properly uploaded to linode bucket when i am uploading suing side attachment button(that is present in all doctypes) but issue is occurring when i am trying to upload files or images through fields of type attach , attach images, text editor.
The files are getting uploaded to linode but the url in the fields is the old url(/publi/… or /private) and same in the case of text editor where the image src of the image is pointing to the /private or /public address.
This is the code in hooks.py
override_doctype_class = {
“File”: “ricemill.linode_storage.CustomFile”
}
This is the code in linode_storage.py
import os
import re
import frappe
from frappe import _
from frappe.utils import now_datetime, get_url, generate_hash,get_site_path
from frappe.utils.password import get_decrypted_password
from datetime import datetime, timedelta
from frappe import enqueue
from frappe.core.doctype.file.file import File as FrappeFile
try:
import boto3
from botocore.client import Config, ClientError
import mimetypes
except ImportError:
frappe.throw(_(“boto3 is required for Linode integration. Install via bench pip install boto3”))
----------------------------------------
1) Helper: Construct a boto3 S3 client
----------------------------------------
def get_linode_client():
“”"
Fetch credentials from Linode Storage Settings and return a boto3 client.
“”"
settings = frappe.get_single(“Linode Storage Settings”)
if not (settings.endpoint_url and settings.access_key and settings.secret_key):
frappe.throw(_(“Please configure Linode Storage Settings first.”))
# Decrypt the password field before passing to boto3
secret_key = get_decrypted_password(
"Linode Storage Settings",
"Linode Storage Settings",
"secret_key"
)
return boto3.client(
"s3",
endpoint_url=settings.endpoint_url,
aws_access_key_id=settings.access_key,
aws_secret_access_key=secret_key,
config=Config(signature_version="s3v4"),
# If you want to fetch the region dynamically, you could store it
# in Linode Storage Settings instead of hard‐coding:
region_name=settings.region or "in-maa-1"
)
-----------------------------------------------------
2) Override the built-in File class so we skip local save
-----------------------------------------------------
class CustomFile(FrappeFile):
“”"
1) We override before_insert so that Frappe does NOT call its default save_file().
Instead, we pull ‘content’, write it to a tmp disk file ourselves, and set file_url
to that disk path—so that validate()/exists_on_disk() sees a valid file path.
2) We override validate_file_on_disk() and validate() to be no-ops, because we
know file_url will be a valid disk path at insert time.
3) In after_insert, we take that disk path, upload to Linode, set file_url to
the Linode URL (public or presigned), delete the local disk file, and insert
FileMetadata.
"""
def get_full_path(self):
"""
Return the absolute filesystem path for this File based on file_url.
By wrapping frappe.get_site_path(file_path) in os.path.abspath(),
we guarantee an absolute path like:
/home/.../frappe-bench/sites/dev.lakshmifoods.org/private/files/xyz.png
rather than a relative string.
"""
# file_url is stored as something like "/private/files/xyz.png"
file_path = (self.file_url or "").lstrip("/") # e.g. "private/files/xyz.png"
# frappe.get_site_path("private/files/xyz.png") might return "sites/dev.lakshmifoods.org/private/files/xyz.png"
# Wrapping in abspath() forces it into the real absolute path on disk.
return os.path.abspath(frappe.get_site_path(file_path))
def before_insert(self):
# 0) If we already handled a write, skip
if getattr(self, "_fm_written", False):
return
# 1) If self.content is not provided (i.e. attach‐field path), skip writing—
# Frappe has already written the file to disk at self.file_url.
if not self.get("content"):
# Mark as “written” so after_insert() still runs exactly once
self._fm_written = True
return
# 2) Otherwise, self.content holds Base64‐encoded bytes. Write them to disk.
try:
raw_bytes = self.get_content() # returns bytes
except Exception as e:
frappe.throw(_("Could not read uploaded content: {0}").format(e))
# 3) Build a random filename to avoid collisions
name_part, ext = os.path.splitext(self.file_name or "")
rand_hash = generate_hash(length=8)
disk_name = f"{name_part}_{rand_hash}{ext}"
# 4) Decide private or public folder
folder = "private/files" if self.is_private else "public/files"
site_path = frappe.get_site_path()
full_dir = os.path.join(site_path, folder)
os.makedirs(full_dir, exist_ok=True)
local_path = os.path.join(full_dir, disk_name)
# 5) Write raw_bytes to disk
try:
with open(local_path, "wb") as f:
f.write(raw_bytes)
except Exception as e:
frappe.throw(_("Could not write file to disk: {0}").format(e))
# 6) Point file_url at the new disk path so that exists_on_disk() will pass
rel_path = os.path.join(folder, disk_name).replace("\\", "/")
self.file_url = "/" + rel_path
# 7) Mark as “written” so we do not do this again
self._fm_written = True
def validate_file_on_disk(self):
# 1) check allowed extensions
name, ext = os.path.splitext(self.file_name)
allowed = frappe.get_conf().file_allowed_extensions or ["txt", "pdf", "jpg", "jpeg", "png", "gif", "webp", "svg", "doc", "docx", "xls", "xlsx", "ppt", "pptx", "csv", "tar", "gz", "mp3", "wav", "mp4", "avi", "mov"]
if ext.lstrip(".").lower() not in allowed:
frappe.throw(_("Files with extension {0} are not allowed.").format(ext))
# 2) file size check
file_size = self.file_size or 0
max_mb = frappe.get_conf().max_file_size_mb or 100
if file_size > max_mb * 1024 * 1024:
frappe.throw(_("Upload exceeds maximum allowed file size of {0} MB.").format(max_mb))
# 3) sanitize the filename
safe_name = re.sub(r'[^a-zA-Z0-9_.\-]', "_", self.file_name or "")
self.file_name = safe_name
# 4) core checks
super(CustomFile, self).validate_file_on_disk()
def validate(self):
# Skip all validation that depends on file_url pointing somewhere else.
return
def after_insert(self):
old_local_url = self.file_url
# frappe.log_error(f"old_local_url: {old_local_url}", "Linode Storage Debug")
if self.custom__linode_handled:
return
# 1a) If file_url already points to an HTTP/S3 URL, skip
if self.file_url and self.file_url.startswith("http"):
frappe.db.set_value("File", self.name, "custom__linode_handled", 1)
return
# # 1) Avoid re-running
# if getattr(self, "_s3_handled", False):
# return
# 1b) If local file is missing, it means this File was created by attaching a previously‐uploaded File. We need to find the ORIGINAL File record whose file_url matched this same local path, then copy its S3 info.
local_path = self.get_full_path()
# frappe.log_error(f"local_path: {local_path}", "Linode Storage Debug")
# frappe.log_error(f"local_path exists: {os.path.exists(local_path)}", "Linode Storage Debug")
if frappe.db.exists("FileMetadata", {"original_url": old_local_url}):
settings = frappe.get_single("Linode Storage Settings")
# frappe.log_error(f"original_file_url field000: {old_local_url}", "Linode Storage Debug")
original_meta = frappe.get_all(
"FileMetadata",
filters={"original_url": old_local_url},
limit_page_length=1,
)
# frappe.log_error(f"original_meta: {original_meta}", "Linode Storage Debug")
if original_meta:
m = frappe.get_doc("FileMetadata", original_meta[0].name)
if m.private:
new_url = f"/api/method/ricemill.linode_storage.get_presigned_download?file_id={m.file_doc}"
frappe.db.set_value("File", self.name, "file_url", new_url)
frappe.db.set_value("File", self.name, "is_private", 1)
# frappe.log_error(f"$$$1$$$$", "Linode Storage Debug")
else:
endpoint = settings.endpoint_url.rstrip("/")
new_url = f"{endpoint}/{m.bucket}/{m.s3_key}"
frappe.db.set_value("File", self.name, "file_url", new_url)
frappe.db.set_value("File", self.name, "is_private", 0)
# frappe.log_error(f"$$$2$$$$", "Linode Storage Debug")
parent_doctype = self.attached_to_doctype
parent_name = self.attached_to_name
if parent_doctype and parent_name and frappe.db.exists(parent_doctype, parent_name):
try:
parent_doc = frappe.get_doc(parent_doctype, parent_name)
# frappe.log_error(f"parent_doc: {parent_doc}", "Linode Storage Debug")
for df in parent_doc.meta.fields:
if df.fieldtype in ("Attach", "Attach Image"):
if parent_doc.get(df.fieldname) == old_local_url:
# frappe.log_error(f"Inside for loop after *****2***", "Linode Storage Debug")
frappe.db.set_value(
parent_doctype, parent_name, df.fieldname, new_url, update_modified=False
)
break
except Exception as e:
frappe.log_error(f"Error updating parent doc: {e}", "Linode Storage Error")
# In all cases, mark as handled and exit
frappe.db.set_value("File", self.name, "custom__linode_handled", 1)
return
# 2) Find FileStorageConfig for this doctype
parent_doctype = self.attached_to_doctype
cfg_list = frappe.get_all(
"FileStorageConfig",
filters={"doctype_name": parent_doctype, "active": 1},
limit_page_length=1
)
if not cfg_list:
frappe.log_error(f"No FileStorageConfig for doctype {parent_doctype}", "Linode Storage Warning")
# No config → we leave the disk file alone and abort
return
config = frappe.get_doc("FileStorageConfig", cfg_list[0].name)
bucket_type = config.bucket_type.strip().lower() # "public" or "private"
folder_name = (config.folder_name or "").strip("/").rstrip("/")
retention_days = config.retention_days or 0
# 3) Build a unique S3 key: "folder_name/<timestamp>_<original_filename>"
timestamp = now_datetime().strftime("%Y%m%d%H%M%S")
# Determine original filename (use file_name if available; otherwise use the basename of file_url)
orig_name = self.file_name or os.path.basename(self.file_url or "")
name_part, ext = os.path.splitext(orig_name.replace(" ", "_"))
unique_filename = f"{timestamp}_{name_part}{ext}"
filename_for_metadata = orig_name or unique_filename
s3_key = f"{folder_name}/{unique_filename}" if folder_name else unique_filename
is_private = (bucket_type == "private")
settings = frappe.get_single("Linode Storage Settings")
s3_client = get_linode_client()
upload_success = False
metadata_success = False
# 4) Upload to Linode
try:
content_type, _ = mimetypes.guess_type(local_path)
if not content_type:
content_type = "application/octet-stream"
extra_args = {}
if not is_private:
extra_args["ACL"] = "public-read"
extra_args["ContentType"] = content_type
extra_args["ContentDisposition"] = "inline"
s3_client.upload_file(
Filename = local_path,
Bucket = (settings.private_bucket if is_private else settings.public_bucket),
Key = s3_key,
ExtraArgs = extra_args
)
upload_success = True
except ClientError as e:
frappe.log_error(message=e, title="Linode Upload Failed")
return
# 5) Update File doc’s URL & is_private flag
if not is_private:
endpoint = settings.endpoint_url.rstrip("/")
public_url = f"{endpoint}/{(settings.public_bucket)}/{s3_key}"
frappe.db.set_value("File", self.name, "file_url", public_url)
frappe.db.set_value("File", self.name, "is_private", 0)
new_http_url = public_url
else:
download_route = f"/api/method/ricemill.linode_storage.get_presigned_download?file_id={self.name}"
frappe.db.set_value("File", self.name, "file_url", download_route)
frappe.db.set_value("File", self.name, "is_private", 1)
new_http_url = download_route
# 6) Insert FileMetadata if it doesn’t exist yet
try:
if not frappe.db.exists("FileMetadata", {"file_doc": self.name}):
meta = frappe.get_doc({
"doctype":"FileMetadata",
"file_doc": self.name,
"attached_to_doctype": self.attached_to_doctype,
"attached_to_name":self.attached_to_name,
"filename": filename_for_metadata,
"bucket": (settings.private_bucket if is_private else settings.public_bucket),
"folder": folder_name,
"s3_key": s3_key,
"uploaded_on": now_datetime(),
"retention_days": retention_days,
"private": (1 if is_private else 0),
"original_url": old_local_url
})
meta.insert(ignore_permissions=True)
metadata_success = True
frappe.db.set_value("File", self.name, "custom__linode_handled", 1)
frappe.db.commit()
# 7) Text editor type fields: replace old local URL with new Linode URL
frappe.enqueue(
"ricemill.linode_storage.replace_url_in_parent", queue='default',
file_name=self.name,
file_url=old_local_url,
attached_to_doctype=self.attached_to_doctype,
attached_to_name=self.attached_to_name,
)
except frappe.DuplicateEntryError:
metadata_success = True
except Exception as e:
frappe.log_error(message=e, title="FileMetadata creation failed")
metadata_success = False
# 8) Mark as handled
if upload_success and metadata_success:
frappe.log_error(f"File {filename_for_metadata} successfully uploaded to Linode Storage", "Linode Storage Success")
# self._s3_handled = True
frappe.log_error(f"File successfully uploaded to Linode Storage", "Linode Storage Success")
def after_update(self):
return self.after_insert()
----------------------------------------
3) Whitelisted endpoint: presigned download
----------------------------------------
@frappe.whitelist(allow_guest=False)
def get_presigned_download(file_id):
frappe.log_error(f"get_presigned_download called for file_id: {file_id}“, “Linode Storage Debug”)
“””
Whitelisted endpoint for downloading a private file.
1. Verify user has read permission on the parent document.
2. Look up FileMetadata by file_doc = file_id.
3. Generate a presigned URL for (bucket, s3_key).
4. Redirect the browser to that presigned URL.
“”"
file_doc = frappe.get_doc(“File”, file_id)
# Permission check: ensure user can read attached document if any
if file_doc.attached_to_doctype and file_doc.attached_to_name:
if not frappe.has_permission(file_doc.attached_to_doctype, "read", file_doc.attached_to_name):
frappe.throw(_("Not permitted to download this file"), frappe.PermissionError)
meta_list = frappe.get_all(
"FileMetadata",
filters={"file_doc": file_id},
limit_page_length=1
)
if not meta_list:
frappe.throw(("FileMetadata not found"), frappe.DoesNotExistError)
meta = frappe.get_doc("FileMetadata", meta_list[0].name)
key = meta.s3_key
settings = frappe.get_single("Linode Storage Settings")
s3 = get_linode_client()
try:
# Determine content type for the object, so the browser can render it inline
content_type, _ = mimetypes.guess_type(meta.filename)
if not content_type:
content_type = "application/octet-stream"
presigned = s3.generate_presigned_url(
ClientMethod='get_object',
Params={
'Bucket': meta.bucket,
'Key': key,
# Force inline display (not download)
'ResponseContentDisposition': f'inline; filename="{meta.filename}"',
# Tell the browser what kind of file this is
'ResponseContentType': content_type
},
ExpiresIn=settings.presign_expiry or 3600
)
except ClientError as e:
frappe.log_error(message=e, title="Error generating presigned URL")
frappe.throw(_("Could not generate presigned URL"))
frappe.local.response["type"] = "redirect"
frappe.local.response["location"] = presigned
Fix-broken image links in text editor fields
def replace_url_in_parent(file_name,file_url, attached_to_doctype, attached_to_name):
frappe.log_error(f"Running replace_url_in_parent for {file_url}", "Linode Debug")
settings = frappe.get_single("Linode Storage Settings")
original_meta = frappe.get_all(
"FileMetadata",
filters={"file_doc": file_name},
limit_page_length=1,
)
frappe.log_error(f"Fetched Metadata: {original_meta}", "Linode Debug")
if original_meta:
m = frappe.get_doc("FileMetadata", original_meta[0].name)
if m.private:
new_url = f"/api/method/ricemill.linode_storage.get_presigned_download?file_id={m.file_doc}"
else:
endpoint = settings.endpoint_url.rstrip("/")
new_url = f"{endpoint}/{m.bucket}/{m.s3_key}"
parent_doc = frappe.get_doc(attached_to_doctype, attached_to_name)
changed = False
for df in parent_doc.meta.fields:
if df.fieldtype in ("Text Editor", "Small Text", "Text", "Code"):
orig_html = getattr(parent_doc, df.fieldname) or ""
pattern = re.escape(f"{file_url}?fid={m.file_doc}")
new_html = re.sub(pattern, new_url, orig_html)
frappe.log_error(f"orig_html for {df.fieldname}=\n{orig_html}", "Linode HTML Debug")
frappe.log_error(f"new_html for {df.fieldname}=\n{new_html}", "Linode HTML Debug")
if new_html != orig_html:
setattr(parent_doc, df.fieldname, new_html)
changed = True
frappe.log_error(f"Changed: {changed}", "Linode Debug")
if changed:
parent_doc.save(ignore_permissions=True)
Can someone help me with this issue ?
is the method of uploading files to linode okay or are there better way of handling this.