blob: 768fd80a6f2f55d59ab5aa779378699cf062b40e [file] [log] [blame]
# Copyright 2014 the Melange authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Utility functions for file operations."""
import cloudstorage
import re
from google.appengine.api import app_identity
from google.appengine.ext import blobstore
# Multipart/form-data forms which want to persist the uploaded files in
# Cloud Storage should have an additional hidden field whose name is equal
# to the string below.
# Existence of this field will prevent middleware from pre-processing uploaded
# files. This part does not apply to uploads which are not handled by Blobstore.
# This constant and corresponding form fields can be removed when all uploads
# are handled Cloud Storage instead of Blobstore.
DONT_USE_BLOBSTORE_MIDDLEWARE = 'dont_use_blobstore_middleware'
def saveFile(file_object, object_name=None, bucket_name=None, folder_name=None):
"""Saves content of the specified file to Cloud Storage.
Args:
file_object: File whose content is saved.
object_name: Name of the object to be saved.
bucket_name: Optional name of the bucket in which the saved object is
supposed to be saved. If not specified, the default bucket for
the application is used.
folder_name: Optional folder name to be prefixed to the path
Returns:
A string containing path under which the object was saved in Cloud Storage.
"""
bucket_name = bucket_name or app_identity.get_default_gcs_bucket_name()
if folder_name:
object_name = '/'.join([folder_name, object_name])
object_path = '/' + '/'.join([bucket_name, object_name])
options = {'x-goog-acl': 'public-read'}
with cloudstorage.open(object_path, 'w', content_type='text/plain',
options=options) as gcs_file:
gcs_file.write(file_object.read())
return object_path
def deleteFile(object_path):
"""Deletes the specified file object from Cloud Storage.
Args:
object_path: The full GCS file name for the object, in the format
/bucket/object_name. Must be a full filename and can include the
delimiter '/'.
"""
try:
cloudstorage.delete(object_path)
except cloudstorage.NotFoundError:
pass
def getBlobKey(object_path):
"""Returns blob key associated with the specified Cloud Storage object.
Args:
object_path: The full GCS file name for the object, in the format
/bucket/object_name. Must be a full filename and can include the initial
delimiter '/'.
Returns:
blobstore.BlobKey associated with the given object.
"""
# strip the leading delimiter if present
if object_path.startswith('/'):
object_path = object_path[1:]
if not object_path:
raise ValueError('Empty path is not accepted.')
full_path = blobstore.GS_PREFIX + object_path
return blobstore.BlobKey(blobstore.create_gs_key(full_path))
SAFE_FILENAME_START_CHARS = r'a-zA-Z0-9_'
SAFE_FILENAME_REST_CHARS = SAFE_FILENAME_START_CHARS + r' ,._-'
SAFE_FILENAME_REGEXP = re.compile(r'^[%s][%s]+$' % (
SAFE_FILENAME_START_CHARS, SAFE_FILENAME_REST_CHARS))
def cleanFilename(filename):
"""Replace unwanted characters in a filename with _.
Args:
filename - a string representing a filename
Returns:
the post-processed filename. If None is passed in, will return None.
"""
if not filename:
return filename
else:
return (re.sub(r'[^%s]' % SAFE_FILENAME_START_CHARS, '_', filename[0]) +
re.sub(r'[^%s]' % SAFE_FILENAME_REST_CHARS, '_', filename[1:]))