refactor: File APIs
Restructured and moved most APIs under frappe.core.api.file namespace.
Changed some obvious security gaps (like using get_list instead of
get_all for an endpoint), styled, added type hints and made minor performance
enhancements.
Changes
* download_file API
* Move API to handler.py
* Check for permissions via File.is_downloadable instead
* Moved APIs to new namespace: `frappe.core.api.file`
* Backwards compatibility
* Added APIs to override_whitelisted_methods to maintain existing
client endpoints
* Imported APIs to controller's namespace to avoid breaking external
app usages
This commit is contained in:
Gavin D'souza
parent
aaf50d28ee
commit
59e45a2e2f
5 changed files with 172 additions and 137 deletions
0
frappe/core/api/__init__.py
Normal file
0
frappe/core/api/__init__.py
Normal file
123
frappe/core/api/file.py
Normal file
123
frappe/core/api/file.py
Normal file
|
|
@ -0,0 +1,123 @@
|
|||
import json
|
||||
from typing import Dict, List
|
||||
|
||||
import frappe
|
||||
from frappe.utils import cint, cstr
|
||||
|
||||
from frappe.core.doctype.file.file import File, setup_folder_path
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def unzip_file(name: str):
|
||||
"""Unzip the given file and make file records for each of the extracted files"""
|
||||
file: File = frappe.get_doc("File", name)
|
||||
return file.unzip()
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def get_attached_images(doctype: str, names: List[str]) -> frappe._dict:
|
||||
"""get list of image urls attached in form
|
||||
returns {name: ['image.jpg', 'image.png']}"""
|
||||
|
||||
if isinstance(names, str):
|
||||
names = json.loads(names)
|
||||
|
||||
img_urls = frappe.db.get_list(
|
||||
"File",
|
||||
filters={
|
||||
"attached_to_doctype": doctype,
|
||||
"attached_to_name": ("in", names),
|
||||
"is_folder": 0,
|
||||
},
|
||||
fields=["file_url", "attached_to_name as docname"],
|
||||
)
|
||||
|
||||
out = frappe._dict()
|
||||
for i in img_urls:
|
||||
out[i.docname] = out.get(i.docname, [])
|
||||
out[i.docname].append(i.file_url)
|
||||
|
||||
return out
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def get_files_in_folder(folder: str, start: int = 0, page_length: int = 20) -> Dict:
|
||||
start = cint(start)
|
||||
page_length = cint(page_length)
|
||||
|
||||
attachment_folder = frappe.db.get_value(
|
||||
"File",
|
||||
"Home/Attachments",
|
||||
["name", "file_name", "file_url", "is_folder", "modified"],
|
||||
as_dict=1,
|
||||
)
|
||||
|
||||
files = frappe.get_list(
|
||||
"File",
|
||||
{"folder": folder},
|
||||
["name", "file_name", "file_url", "is_folder", "modified"],
|
||||
start=start,
|
||||
page_length=page_length + 1,
|
||||
)
|
||||
|
||||
if folder == "Home" and attachment_folder not in files:
|
||||
files.insert(0, attachment_folder)
|
||||
|
||||
return {"files": files[:page_length], "has_more": len(files) > page_length}
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def get_files_by_search_text(text: str) -> List[Dict]:
|
||||
if not text:
|
||||
return []
|
||||
|
||||
text = "%" + cstr(text).lower() + "%"
|
||||
return frappe.get_list(
|
||||
"File",
|
||||
fields=["name", "file_name", "file_url", "is_folder", "modified"],
|
||||
filters={"is_folder": False},
|
||||
or_filters={
|
||||
"file_name": ("like", text),
|
||||
"file_url": text,
|
||||
"name": ("like", text),
|
||||
},
|
||||
order_by="modified desc",
|
||||
limit=20,
|
||||
)
|
||||
|
||||
|
||||
@frappe.whitelist(allow_guest=True)
|
||||
def get_max_file_size() -> int:
|
||||
return cint(frappe.conf.get("max_file_size")) or 10485760
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def create_new_folder(file_name: str, folder: str) -> File:
|
||||
"""create new folder under current parent folder"""
|
||||
file = frappe.new_doc("File")
|
||||
file.file_name = file_name
|
||||
file.is_folder = 1
|
||||
file.folder = folder
|
||||
file.insert(ignore_if_duplicate=True)
|
||||
return file
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def move_file(file_list: List[File], new_parent: str, old_parent: str) -> None:
|
||||
if isinstance(file_list, str):
|
||||
file_list = json.loads(file_list)
|
||||
|
||||
for file_obj in file_list:
|
||||
setup_folder_path(file_obj.get("name"), new_parent)
|
||||
|
||||
# recalculate sizes
|
||||
frappe.get_doc("File", old_parent).save()
|
||||
frappe.get_doc("File", new_parent).save()
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def zip_files(files: str):
|
||||
files = frappe.parse_json(files)
|
||||
frappe.response["filename"] = "files.zip"
|
||||
frappe.response["filecontent"] = File.zip_files(files)
|
||||
frappe.response["type"] = "download"
|
||||
|
|
@ -2,11 +2,11 @@
|
|||
# License: MIT. See LICENSE
|
||||
|
||||
import io
|
||||
import json
|
||||
import mimetypes
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
from typing import List
|
||||
import zipfile
|
||||
|
||||
from requests.exceptions import HTTPError, SSLError
|
||||
|
|
@ -16,9 +16,10 @@ from urllib.parse import quote, unquote
|
|||
import frappe
|
||||
from frappe import _
|
||||
from frappe.model.document import Document
|
||||
from frappe.utils import call_hook_method, cint, cstr, encode, get_files_path, get_hook_method
|
||||
from frappe.utils import call_hook_method, cint, encode, get_files_path, get_hook_method
|
||||
from frappe.utils.image import strip_exif_data, optimize_image
|
||||
from frappe.utils.file_manager import is_safe_path, safe_b64decode
|
||||
from frappe.core.api.file import *
|
||||
|
||||
from .exceptions import MaxFileSizeReachedError, FolderNotEmpty
|
||||
from .utils import *
|
||||
|
|
@ -313,7 +314,7 @@ class File(Document):
|
|||
self.flags.on_rollback = True
|
||||
self.on_trash()
|
||||
|
||||
def unzip(self):
|
||||
def unzip(self) -> List[File]:
|
||||
'''Unzip current file and replace it by its children'''
|
||||
if not self.file_url.endswith(".zip"):
|
||||
frappe.throw(_("{0} is not a zip file").format(self.file_name))
|
||||
|
|
@ -578,44 +579,6 @@ def on_doctype_update():
|
|||
frappe.db.add_index("File", ["attached_to_doctype", "attached_to_name"])
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def create_new_folder(file_name, folder):
|
||||
""" create new folder under current parent folder """
|
||||
file = frappe.new_doc("File")
|
||||
file.file_name = file_name
|
||||
file.is_folder = 1
|
||||
file.folder = folder
|
||||
file.insert(ignore_if_duplicate=True)
|
||||
return file
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def move_file(file_list, new_parent, old_parent):
|
||||
|
||||
if isinstance(file_list, str):
|
||||
file_list = json.loads(file_list)
|
||||
|
||||
for file_obj in file_list:
|
||||
setup_folder_path(file_obj.get("name"), new_parent)
|
||||
|
||||
# recalculate sizes
|
||||
frappe.get_doc("File", old_parent).save()
|
||||
frappe.get_doc("File", new_parent).save()
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def zip_files(files):
|
||||
files = frappe.parse_json(files)
|
||||
frappe.response["filename"] = "files.zip"
|
||||
frappe.response["filecontent"] = File.zip_files(files)
|
||||
frappe.response["type"] = "download"
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def get_max_file_size():
|
||||
return cint(frappe.conf.get('max_file_size')) or 10485760
|
||||
|
||||
|
||||
def has_permission(doc, ptype=None, user=None):
|
||||
has_access = False
|
||||
user = user or frappe.session.user
|
||||
|
|
@ -648,92 +611,3 @@ def has_permission(doc, ptype=None, user=None):
|
|||
pass
|
||||
|
||||
return has_access
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def download_file(file_url):
|
||||
"""
|
||||
Download file using token and REST API. Valid session or
|
||||
token is required to download private files.
|
||||
|
||||
Method : GET
|
||||
Endpoint : frappe.core.doctype.file.file.download_file
|
||||
URL Params : file_name = /path/to/file relative to site path
|
||||
"""
|
||||
file_doc = frappe.get_doc("File", {"file_url": file_url})
|
||||
file_doc.check_permission("read")
|
||||
|
||||
frappe.local.response.filename = os.path.basename(file_url)
|
||||
frappe.local.response.filecontent = file_doc.get_content()
|
||||
frappe.local.response.type = "download"
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def unzip_file(name):
|
||||
'''Unzip the given file and make file records for each of the extracted files'''
|
||||
file_obj: File = frappe.get_doc('File', name)
|
||||
return file_obj.unzip()
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def get_attached_images(doctype, names):
|
||||
'''get list of image urls attached in form
|
||||
returns {name: ['image.jpg', 'image.png']}'''
|
||||
|
||||
if isinstance(names, str):
|
||||
names = json.loads(names)
|
||||
|
||||
img_urls = frappe.db.get_list('File', filters={
|
||||
'attached_to_doctype': doctype,
|
||||
'attached_to_name': ('in', names),
|
||||
'is_folder': 0
|
||||
}, fields=['file_url', 'attached_to_name as docname'])
|
||||
|
||||
out = frappe._dict()
|
||||
for i in img_urls:
|
||||
out[i.docname] = out.get(i.docname, [])
|
||||
out[i.docname].append(i.file_url)
|
||||
|
||||
return out
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def get_files_in_folder(folder, start=0, page_length=20):
|
||||
start = cint(start)
|
||||
page_length = cint(page_length)
|
||||
|
||||
attachment_folder = frappe.db.get_value('File',
|
||||
'Home/Attachments',
|
||||
['name', 'file_name', 'file_url', 'is_folder', 'modified'],
|
||||
as_dict=1
|
||||
)
|
||||
|
||||
files = frappe.db.get_list('File',
|
||||
{ 'folder': folder },
|
||||
['name', 'file_name', 'file_url', 'is_folder', 'modified'],
|
||||
start=start,
|
||||
page_length=page_length + 1
|
||||
)
|
||||
|
||||
if folder == 'Home' and attachment_folder not in files:
|
||||
files.insert(0, attachment_folder)
|
||||
|
||||
return {
|
||||
'files': files[:page_length],
|
||||
'has_more': len(files) > page_length
|
||||
}
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def get_files_by_search_text(text):
|
||||
if not text:
|
||||
return []
|
||||
|
||||
text = '%' + cstr(text).lower() + '%'
|
||||
return frappe.get_all('File',
|
||||
fields=['name', 'file_name', 'file_url', 'is_folder', 'modified'],
|
||||
filters={'is_folder': False},
|
||||
or_filters={'file_name': ('like', text), 'file_url': text, 'name': ('like', text)},
|
||||
order_by='modified desc',
|
||||
limit=20
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,18 +1,24 @@
|
|||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: MIT. See LICENSE
|
||||
|
||||
import os
|
||||
from mimetypes import guess_type
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from werkzeug.wrappers import Response
|
||||
|
||||
import frappe
|
||||
import frappe.utils
|
||||
import frappe.sessions
|
||||
from frappe.utils import cint
|
||||
import frappe.utils
|
||||
from frappe import _, is_whitelisted
|
||||
from frappe.utils.response import build_response
|
||||
from frappe.core.doctype.server_script.server_script_utils import get_server_script_map
|
||||
from frappe.utils import cint
|
||||
from frappe.utils.csvutils import build_csv_response
|
||||
from frappe.utils.image import optimize_image
|
||||
from mimetypes import guess_type
|
||||
from frappe.core.doctype.server_script.server_script_utils import get_server_script_map
|
||||
from frappe.utils.response import build_response
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from frappe.core.doctype.file.file import File
|
||||
|
||||
|
||||
ALLOWED_MIMETYPES = ('image/png', 'image/jpeg', 'application/pdf', 'application/msword',
|
||||
|
|
@ -209,6 +215,25 @@ def upload_file():
|
|||
return ret
|
||||
|
||||
|
||||
@frappe.whitelist(allow_guest=True)
|
||||
def download_file(file_url: str):
|
||||
"""
|
||||
Download file using token and REST API. Valid session or
|
||||
token is required to download private files.
|
||||
|
||||
Method : GET
|
||||
Endpoints : download_file, frappe.core.doctype.file.file.download_file
|
||||
URL Params : file_name = /path/to/file relative to site path
|
||||
"""
|
||||
file: "File" = frappe.get_doc("File", {"file_url": file_url})
|
||||
if not file.is_downloadable():
|
||||
raise frappe.PermissionError
|
||||
|
||||
frappe.local.response.filename = os.path.basename(file_url)
|
||||
frappe.local.response.filecontent = file.get_content()
|
||||
frappe.local.response.type = "download"
|
||||
|
||||
|
||||
def get_attr(cmd):
|
||||
"""get method object from cmd"""
|
||||
if '.' in cmd:
|
||||
|
|
@ -218,6 +243,7 @@ def get_attr(cmd):
|
|||
frappe.log("method:" + cmd)
|
||||
return method
|
||||
|
||||
|
||||
@frappe.whitelist(allow_guest=True)
|
||||
def ping():
|
||||
return "pong"
|
||||
|
|
@ -225,8 +251,8 @@ def ping():
|
|||
|
||||
def run_doc_method(method, docs=None, dt=None, dn=None, arg=None, args=None):
|
||||
"""run a whitelisted controller method"""
|
||||
import json
|
||||
import inspect
|
||||
import json
|
||||
|
||||
if not args:
|
||||
args = arg or ""
|
||||
|
|
|
|||
|
|
@ -383,3 +383,15 @@ global_search_doctypes = {
|
|||
{"doctype": "Web Form"}
|
||||
]
|
||||
}
|
||||
|
||||
override_whitelisted_methods = {
|
||||
"frappe.core.doctype.file.file.download_file": "download_file",
|
||||
"frappe.core.doctype.file.file.unzip_file": "frappe.core.api.file.unzip_file",
|
||||
"frappe.core.doctype.file.file.get_attached_images": "frappe.core.api.file.get_attached_images",
|
||||
"frappe.core.doctype.file.file.get_files_in_folder": "frappe.core.api.file.get_files_in_folder",
|
||||
"frappe.core.doctype.file.file.get_files_by_search_text": "frappe.core.api.file.get_files_by_search_text",
|
||||
"frappe.core.doctype.file.file.get_max_file_size": "frappe.core.api.file.get_max_file_size",
|
||||
"frappe.core.doctype.file.file.create_new_folder": "frappe.core.api.file.create_new_folder",
|
||||
"frappe.core.doctype.file.file.move_file": "frappe.core.api.file.move_file",
|
||||
"frappe.core.doctype.file.file.zip_files": "frappe.core.api.file.zip_files",
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue