本文整理汇总了Python中os.path._join函数的典型用法代码示例。如果您正苦于以下问题:Python _join函数的具体用法?Python _join怎么用?Python _join使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了_join函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: page_media_upload
def page_media_upload(page_id):
user = auth.is_logged_in(request)
page = Page.load(page_id)
permission = auth.is_page_editor(user, page)
overwrite = []
for n in request.files:
x = request.files.get(n)
media_path = _join(page.blog.path, page.blog.media_path_generated)
file_path = _join(media_path, x.filename)
if _exists(file_path):
from core.error import FileExistsError
raise FileExistsError("File '{}' already exists on the server.".format(
utils.html_escape(x.filename)))
else:
Media.register_media(x.filename, file_path, user, page=page)
if not _exists(media_path):
makedirs(media_path)
x.save(file_path)
tags = template_tags(page=page)
return template('edit/page_media_list.tpl',
**tags.__dict__)
开发者ID:syegulalp,项目名称:mercury,代码行数:26,代码来源:page.py
示例2: register_plugin
def register_plugin(path_to_plugin, **ka):
if os.path.isfile(_join(PLUGIN_PATH, path_to_plugin, "__init__.py")):
try:
added_plugin = importlib.import_module("data.plugins." + path_to_plugin)
except SystemError:
raise PluginImportError("Plugin at " +
_join(PLUGIN_PATH, path_to_plugin) + " could not be registered.")
else:
try:
existing_plugin = Plugin.select().where(
Plugin.path == path_to_plugin).get()
except Plugin.DoesNotExist:
new_plugin = Plugin(
name=path_to_plugin,
friendly_name=added_plugin.__plugin_name__,
path=path_to_plugin,
priority=1,
enabled=ka.get('enable', False)
)
new_plugin.save()
plugin_data = added_plugin.install()
try:
plugin_settings = plugin_data.get(['settings'])
except (AttributeError, TypeError):
pass
except Exception as e:
raise e
else:
from core.models import PluginData
for n in plugin_settings:
# TODO: instead: iter through __dict__
# if dict item not in field list, don't add
settings_data = PluginData(
plugin=new_plugin,
key=n.get('key'),
text_value=n.get('text_value'),
int_value=n.get('int_value'),
blog=n.get('blog'),
site=n.get('site'),
parent=n.get('parent')
)
settings_data.save()
_stddebug ("Plugin registered: " + added_plugin.__plugin_name__ + "\n")
return new_plugin
else:
raise PluginImportError("Plugin at " + PLUGIN_FILE_PATH +
"/" + path_to_plugin + " is already registered.")
开发者ID:syegulalp,项目名称:mercury,代码行数:57,代码来源:plugins.py
示例3: deploy
def deploy(branch):
"""
Deploy the application in a timestamped release folder.
$ fab deploy:staging
Internally this does the following:
* `git pull` if a cached repository already exists
* `git clone` if it's the first deploy ever
* Checkout the current selected branch
* Create a new timestamped release directory
* Copy the cached repository to the new release directory
* Setup the virtualenv
* Install PIP's requirements, downloading new ones if not already cached
* Symlink `<branch>/current` to `<branch>/releases/<timestamped release directory>`
"""
if not git.is_repository(_repo_path(env.github_repo_name)):
# repository doesn't exist, do a fresh clone
with cd(env.repo_path):
git.clone(env.github_repo, env.github_repo_name)
with _repo(env.github_repo_name):
git.checkout(branch)
else:
# repository exists
with _repo(env.github_repo_name):
if not (branch == git.current_branch()):
# switch to our branch if not already
git.checkout(branch)
# pull in the latest code
git.pull(branch)
# 20100603_125848
new_release_name = datetime.utcnow().strftime(RELEASE_NAME_FORMAT)
# /var/praekelt/richmond/staging/releases/20100603_125848
new_release_path = _join(env.releases_path, new_release_name)
# /var/praekelt/richmond/staging/releases/20100603_125848/richmond
# Django needs the project name as it's parent dir since that is
# automagically appended to the loadpath
new_release_repo = _join(new_release_path, env.github_repo_name)
system.create_dir(new_release_path)
system.copy_dirs(_repo_path(env.github_repo_name), new_release_path)
copy_settings_file(branch, release=new_release_name)
symlink_shared_dirs = ['logs', 'tmp']
for dirname in symlink_shared_dirs:
with cd(new_release_repo):
system.remove(dirname, recursive_force=True)
system.symlink(_join(env.shared_path, dirname), dirname)
# create the virtualenv
create_virtualenv(branch)
# ensure we're deploying the exact revision as we locally have
base.set_current(new_release_name)
开发者ID:euan,项目名称:richmond,代码行数:56,代码来源:fabfile.py
示例4: activate_plugins
def activate_plugins():
_stddebug("Activating plugins.\n")
plugin_errors = []
plugins_to_activate = Plugin.select().where(Plugin.enabled == True)
for n in plugins_to_activate:
try:
added_plugin = importlib.import_module("data.plugins." + n.path)
except ImportError as e:
plugin_errors.append("\nPlugin " + n.friendly_name +
" could not be activated. The path '" + _join(PLUGIN_FILE_PATH, n.path) +
"' may be wrong. ({})".format(str(e)))
continue
except SystemError as e:
plugin_errors.append("\nPlugin at '" + _join(PLUGIN_FILE_PATH , n.path) +
"' could not be activated. The plugin may be improperly installed.".format(e))
continue
try:
for m in plugin_attributes:
p_a = added_plugin.__getattribute__(m)
except AttributeError as e:
plugin_errors.append("\nPlugin at '" + _join(PLUGIN_FILE_PATH , n.path) +
"' is missing one or more of its configuration attributes. The plugin may be damaged or improperly installed. ({})".format(e))
continue
plugin_list[n.id] = added_plugin
try:
plugin_loader = added_plugin.load()
for func in plugin_loader:
action = plugin_action[func['action']]
module = importlib.import_module(func['module'])
func_to_wrap = module.__dict__[func['function']]
if action == 'exec':
func_to_wrap(**func['kwargs'])
else:
func_wrapper = func['wrap']
module.__dict__[func['function']] = action(func_wrapper)(func_to_wrap)
except BaseException as e:
plugin_errors.append("\nPlugin at '" + _join(PLUGIN_FILE_PATH, n.path) +
"' could not be activated. Its source may be damaged. ({})".format(e))
continue
_stddebug("Plugin activated: " + added_plugin.__plugin_name__ + "\n")
if len(plugin_errors) > 0:
raise PluginImportError(''.join(plugin_errors))
_stddebug("\n")
开发者ID:syegulalp,项目名称:mercury,代码行数:55,代码来源:plugins.py
示例5: f_read
def f_read(f_name):
ret = ""
try:
with open(_join(_expanduser("~"), ".klb/" + f_name)) as f:
ret += f.read()
except:
pass
if _getenv("SLAVE"):
try:
with open(_join(_expanduser("~"), ".slave/" + f_name)) as f:
ret += f.read()
except:
pass
return ret
开发者ID:Giannie,项目名称:KLBFAX,代码行数:14,代码来源:file_handler.py
示例6: supervisor
def supervisor(branch, command):
"""Issue a supervisord command"""
app_path = _join(env.current, env.github_repo_name)
pid_path = _join(app_path,"tmp","pids","supervisord.pid")
if not exists(pid_path):
_virtualenv(
app_path,
"supervisord -c supervisord.%s.conf -j %s" % (branch,pid_path)
)
_virtualenv(
_join(env.current, env.github_repo_name),
"supervisorctl -c supervisord.%s.conf %s" % (branch, command)
)
开发者ID:dmaclay,项目名称:vumi,代码行数:14,代码来源:fabfile.py
示例7: update_io_registry
def update_io_registry(wkdir, mpid, iocomp_types=None):
"""helper method to correctly update the IoRegistry instances
"""
import os
from os.path import join as _join
from os.path import basename as _basename
from os.path import isabs as _isabs
from PyUtils.PoolFile import PoolFileCatalog
# ioreg is a dict:
# {'iocomp-name' : { 'old-fname' : ['iomode', 'new-fname'] }, ... }
ioreg = IoRegistry.instances
msg.debug("ioreg::: %s" % ioreg)
pfc = PoolFileCatalog()
ioreg_items = IoRegistry.instances.iteritems()
for iocomp,iodata in ioreg_items:
#print "--iocomp,len(iodata)",iocomp, len(iodata)
io_items = iodata.iteritems()
for ioname,ioval in io_items:
# handle logical filenames...
#ioname=pfc(ioname)
pfc_name = pfc(ioname)
if (pfc_name != ioname):
ioreg[iocomp][ioname][1]=pfc_name
##print " --iocomp,ioname,ioval",iocomp,ioname,ioval
iomode,newname = ioval[0], ioval[1] or ioname
if iomode == '<output>':
newname = _join (wkdir,
"mpid_%s__%s"%(str(mpid).zfill(3),
_basename(ioname)))
msg.debug ("update_io_registry:<output>: newname=%s" % newname)
elif iomode == '<input>':
if not _isabs(ioname) and not ioname.startswith("root:") and not ioname.startswith("rfio"):
# FIXME: handle URLs/URIs...
src = os.path.abspath(_join(os.curdir, ioname))
dst = _join(wkdir, ioname)
os.symlink(src, dst)
msg.debug( "update_io_registry:<input> created symlink %s for" % dst)
else:
raise ValueError, "unexpected iomode value: %r"%iomode
ioreg[iocomp][ioname][1] = newname
pass
pass
msg.debug( "IoRegistry.instances=%s" % IoRegistry.instances )
return # update_io_registry
开发者ID:atlas-control,项目名称:control,代码行数:49,代码来源:IoUtils.py
示例8: __init__
def __init__(self, globalWorkingDir, localWorkingDir, outputs=None, job=None, esJobManager=None, outputDir=None, rank=None, logger=None):
threading.Thread.__init__(self)
self.__globalWorkingDir = globalWorkingDir
self.__localWorkingDir = localWorkingDir
self.__currentDir = None
self.__rank = rank
if logger and False:
self.__tmpLog = logger
else:
curdir = _abspath (self.__localWorkingDir)
wkdirname = "rank_%s" % str(self.__rank)
wkdir = _abspath (_join(curdir,wkdirname))
self.__tmpLog = Logger.Logger(filename=os.path.join(wkdir, 'Droid.log'))
self.__job = job
self.__esJobManager = esJobManager
self.__stop = threading.Event()
self.__isFinished = False
self.__tmpLog.info("Rank %s: Global working dir: %s" % (self.__rank, self.__globalWorkingDir))
os.environ['PilotHomeDir'] = os.path.dirname(self.__globalWorkingDir)
self.__jobId = None
self.__copyOutputToGlobal = False
self.__outputDir = outputDir
self.__hostname = socket.getfqdn()
self.__outputs = outputs
self.__threadpool = None
self.setup(job)
开发者ID:PanDAWMS,项目名称:pilot,代码行数:29,代码来源:DroidStager.py
示例9: f_readlines
def f_readlines(f_name):
ret = []
try:
with open(_join(_expanduser("~"), ".klb/" + f_name)) as f:
ret += f.readlines()
except:
pass
if _getenv("SLAVE"):
try:
with open(_join(_expanduser("~"), ".slave/" + f_name)) as f:
ret += f.readlines()
except:
pass
for i in range(len(ret)):
ret[i] = ret[i].strip("\n")
return ret
开发者ID:Giannie,项目名称:KLBFAX,代码行数:16,代码来源:file_handler.py
示例10: page_media_upload_confirm
def page_media_upload_confirm(page_id):
user = auth.is_logged_in(request)
page = Page.load(page_id)
permission = auth.is_page_editor(user, page)
# get file NAMES, attributes, size, etc. first
# request.form.getunicode('filename')
# check each one on the SERVER side, not the client
# if each file is OK, then respond appropriately and have the client send the whole file
# if not, respond with a warning to be added to the notification area
_g = request.forms.getunicode
file_name = _g('filename')
file_size = _g('filesize')
# check for file types against master list
# check for file length
# check for name collision
for n in request.files:
x = request.files.get(n)
file_path = _join(page.blog.path, page.blog.media_path_generated, x.filename)
if _exists(file_path):
pass
else:
pass
开发者ID:syegulalp,项目名称:mercury,代码行数:28,代码来源:page.py
示例11: copy_settings_file
def copy_settings_file(branch, release=None):
"""
Copy the settings file for this branch to the server
$ fab copy_settings_file:staging
If no release is specified it defaults to the latest release.
"""
release = release or base.current_release()
directory = _join(env.releases_path, release, env.github_repo_name)
put(
"environments/%(branch)s.py" % env,
_join(directory, "environments/%(branch)s.py" % env)
)
开发者ID:euan,项目名称:richmond,代码行数:16,代码来源:fabfile.py
示例12: f_read_pickle
def f_read_pickle(f_name, path=None):
if path == None:
path = default_path
try:
with open(_join(path, f_name), "r") as f:
return pickle.load(f)
except:
pass
开发者ID:Giannie,项目名称:KLBFAX,代码行数:8,代码来源:file_handler.py
示例13: __init__
def __init__(self):
db.bind('postgres', **_conf.dbconf)
db.generate_mapping(create_tables=True)
settings = dict(
static_path = _join(_projdir, 'statics'),
static_hash_cache = not _options.debug,
template_path = _join(_projdir, 'templates'),
compile_template_cache = not _options.debug,
#compress_response = True,
cookie_secret = 'NGM0NTRkNDIyZDRiNDg0MTU3NDE1ODNhNDU2YzYxNjM2NTcz',
xsrf_cookies = True,
login_url = '/login',
server_traceback = _options.debug,
debug = _options.debug
)
#print url_handlers
super(App_Server, self).__init__(handlers=_url_handlers, **settings)
开发者ID:enlacescomunitarios,项目名称:enlaces,代码行数:17,代码来源:app_server.py
示例14: rm_miui_res
def rm_miui_res():
if _exists(miui_res_dir):
shutil.rmtree(miui_res_dir)
for dir in get_dependency_projects():
dst_dir = _join(dir, miui_res_dir)
if _exists(dst_dir):
shutil.rmtree(dst_dir)
开发者ID:HuangWenhuan0,项目名称:email-build,代码行数:8,代码来源:util.py
示例15: f_write_pickle
def f_write_pickle(f_name, var, path=None):
if path == None:
path = default_path
try:
with open(_join(path, f_name), "w") as f:
pickle.dump(var, f)
except:
pass
开发者ID:Giannie,项目名称:KLBFAX,代码行数:8,代码来源:file_handler.py
示例16: f_read_json
def f_read_json(f_name):
ret = {}
try:
with open(_join(_expanduser("~"), ".klb/" + f_name)) as f:
temp = _json.load(f)
for a in temp:
ret[a] = temp[a]
except:
pass
if _getenv("SLAVE"):
try:
with open(_join(_expanduser("~"), ".slave/" + f_name)) as f:
temp = _json.load(f)
for a in temp:
ret[a] = temp[a]
except:
pass
return ret
开发者ID:Giannie,项目名称:KLBFAX,代码行数:18,代码来源:file_handler.py
示例17: initWorkingDir
def initWorkingDir(self):
# Create separate working directory for each rank
curdir = _abspath (self.localWorkingDir)
wkdirname = "rank_%s" % str(self.rank)
wkdir = _abspath (_join(curdir,wkdirname))
if not os.path.exists(wkdir):
os.makedirs (wkdir)
os.chdir (wkdir)
self.currentDir = wkdir
开发者ID:PanDAWMS,项目名称:pilot,代码行数:9,代码来源:Yoda.py
示例18: restart_webapp
def restart_webapp(branch, **kwargs):
"""
Restart the webapp
$ fab restart_webapp:staging,port=8000
"""
_virtualenv(
_join(env.current, env.github_repo_name),
twistd.restart_command('richmond_webapp', **kwargs)
)
开发者ID:euan,项目名称:richmond,代码行数:11,代码来源:fabfile.py
示例19: list_celery_workers
def list_celery_workers(branch):
"""
List all running celery workers
$ fab list_celery_workers:staging
"""
with cd(_join(env.current, env.github_repo_name)):
sessions = _get_screens("celery_")
for pid,uuid in sessions:
print "Celery Worker => pid:%s, uuid:%s" % (pid, uuid)
开发者ID:euan,项目名称:richmond,代码行数:11,代码来源:fabfile.py
示例20: stop_webapp
def stop_webapp(branch, **kwargs):
"""
Stop the webapp
$ fab stop_webapp:staging,port=8000
"""
_virtualenv(
_join(env.current, env.github_repo_name),
twistd.stop_command('richmond_webapp', **kwargs)
)
开发者ID:euan,项目名称:richmond,代码行数:11,代码来源:fabfile.py
注:本文中的os.path._join函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论