Python os 模块,tmpfile() 实例源码
我们从Python开源项目中,提取了以下15个代码示例,用于说明如何使用os.tmpfile()。
def ProcSeries(cmd,i,njobs,email=False,inName='logIn.log',outName='logOut.log',errName='logErr.log'):
global firstRun
if firstRun:
logIn = open(inName, 'w')
logOut = open(outName, 'w')
logErr = open(errName, 'w')
firstRun = False
else:
logIn = open(inName, 'a')
logOut = open(outName, 'a')
logErr = open(errName, 'a')
inl = os.tmpfile()
outl = os.tmpfile()
errl = os.tmpfile()
print '\033[1;33m-> %s\033[0m [%i of %i]' %(cmd,i+1,njobs)
start = datetime.datetime.now()
if email==False:
proc = call(cmd,shell=True,stdin=inl,stdout=outl,stderr=errl)
else:
proc = call(_Notifier(cmd,i+1,njobs,email),shell=True,stdin=inl,stdout=outl,stderr=errl)
end = datetime.datetime.now()
print 'Process %i of %i completed. Time: %s' %(i+1,njobs,end-start)
_Temp2Perm(inl,logIn,cmd,start,end)
_Temp2Perm(outl,logOut,cmd,start,end)
if errl.tell() != 0:
print '\033[1;31mErrors have occurred!\033[0m (check %s)' %(logErr.name)
_Temp2Perm(errl,logErr,cmd,start,end)
if i==njobs-1:
firstRun = True
def ProcParallel(cmd,i,njobs,email=False,inName='logIn.log',outName='logOut.log',errName='logErr.log'):
global firstRun
global procs
if firstRun:
logIn = open(inName, 'w')
logOut = open(outName, 'w')
logErr = open(errName, 'w')
firstRun = False
else:
logIn = open(inName, 'a')
logOut = open(outName, 'a')
logErr = open(errName, 'a')
inl = os.tmpfile()
outl = os.tmpfile()
errl = os.tmpfile()
print '\033[1;33m-> %s\033[0m [%i of %i]' %(cmd,i+1,njobs)
start = datetime.datetime.now()
if email==False:
proc = Popen(cmd,shell=True,stdin=inl,stdout=outl,stderr=errl)
else:
proc = Popen(_Notifier(cmd,i+1,njobs,email),shell=True,stdin=inl,stdout=outl,stderr=errl)
procs.append((start,cmd,proc,inl,outl,errl))
if i==njobs-1:
for j, (start, name, proc, inl, outl, errl) in enumerate(procs):
proc.wait()
end = datetime.datetime.now()
print 'Process %i of %i completed. Time: %s' %(j+1,njobs,end-start)
_Temp2Perm(inl,logIn,name,start,end)
_Temp2Perm(outl,logOut,name,start,end)
if errl.tell() != 0:
print '\033[1;31mErrors have occurred!\033[0m (check %s)' %(logErr.name)
_Temp2Perm(errl,logErr,cmd,start,end)
firstRun = True
procs = []
def setUp(self):
self.tfile = os.tmpfile()
self.cur_year = datetime.date.today().year
self.cur_year_str = str(self.cur_year)
def setUp(self):
self.tfile = os.tmpfile()
self.cur_year = datetime.date.today().year
self.cur_year_str = str(self.cur_year)
def run(self):
bld = self.generator.bld
linked=[]
target_paths = []
for g in bld.groups:
for tgen in g:
# FIXME it might be better to check if there is a link_task (getattr?)
target_paths += [tgen.path.get_bld().bldpath()]
linked += [t.outputs[0].bldpath()
for t in getattr(tgen, 'tasks', [])
if t.__class__.__name__ in
['cprogram', 'cshlib', 'cxxprogram', 'cxxshlib']]
lib_list = []
if len(linked):
cmd = [self.env.LDD] + linked
# FIXME add DYLD_LIBRARY_PATH+PATH for osx+win32
ldd_env = {'LD_LIBRARY_PATH': ':'.join(target_paths + self.env.LIBPATH)}
# FIXME the with syntax will not work in python 2
with tmpfile() as result:
self.exec_command(cmd, env=ldd_env, stdout=result)
result.seek(0)
for line in result.readlines():
words = line.split()
if len(words) < 3 or words[1] != '=>': continue
lib = words[2]
if lib == 'not': continue
if any([lib.startswith(p) for p in
[bld.bldnode.abspath(), '('] +
self.env.SOFTLINK_EXCLUDE]):
continue
if not isabs(lib):
continue
lib_list.append(lib)
lib_list = sorted(set(lib_list))
self.outputs[0].write(linesep.join(lib_list + self.env.DYNAMIC_LIBS))
return 0
def run(self):
bld = self.generator.bld
linked=[]
target_paths = []
for g in bld.groups:
for tgen in g:
# FIXME it might be better to check if there is a link_task (getattr?)
target_paths += [tgen.path.get_bld().bldpath()]
linked += [t.outputs[0].bldpath()
for t in getattr(tgen, 'tasks', [])
if t.__class__.__name__ in
['cprogram', 'cshlib', 'cxxprogram', 'cxxshlib']]
lib_list = []
if len(linked):
cmd = [self.env.LDD] + linked
# FIXME add DYLD_LIBRARY_PATH+PATH for osx+win32
ldd_env = {'LD_LIBRARY_PATH': ':'.join(target_paths + self.env.LIBPATH)}
# FIXME the with syntax will not work in python 2
with tmpfile() as result:
self.exec_command(cmd, env=ldd_env, stdout=result)
result.seek(0)
for line in result.readlines():
words = line.split()
if len(words) < 3 or words[1] != '=>': continue
lib = words[2]
if lib == 'not': continue
if any([lib.startswith(p) for p in
[bld.bldnode.abspath(), '('] +
self.env.SOFTLINK_EXCLUDE]):
continue
if not isabs(lib):
continue
lib_list.append(lib)
lib_list = sorted(set(lib_list))
self.outputs[0].write(linesep.join(lib_list + self.env.DYNAMIC_LIBS))
return 0
def run(self):
bld = self.generator.bld
linked=[]
target_paths = []
for g in bld.groups:
for tgen in g:
# FIXME it might be better to check if there is a link_task (getattr?)
target_paths += [tgen.path.get_bld().bldpath()]
linked += [t.outputs[0].bldpath()
for t in getattr(tgen, 'tasks', [])
if t.__class__.__name__ in
['cprogram', 'cshlib', 'cxxprogram', 'cxxshlib']]
lib_list = []
if len(linked):
cmd = [self.env.LDD] + linked
# FIXME add DYLD_LIBRARY_PATH+PATH for osx+win32
ldd_env = {'LD_LIBRARY_PATH': ':'.join(target_paths + self.env.LIBPATH)}
# FIXME the with syntax will not work in python 2
with tmpfile() as result:
self.exec_command(cmd, env=ldd_env, stdout=result)
result.seek(0)
for line in result.readlines():
words = line.split()
if len(words) < 3 or words[1] != '=>': continue
lib = words[2]
if lib == 'not': continue
if any([lib.startswith(p) for p in
[bld.bldnode.abspath(), '('] +
self.env.SOFTLINK_EXCLUDE]):
continue
if not isabs(lib):
continue
lib_list.append(lib)
lib_list = sorted(set(lib_list))
self.outputs[0].write(linesep.join(lib_list + self.env.DYNAMIC_LIBS))
return 0
def patch_seq2ropath(patch_seq):
"""Apply the patches in patch_seq, return single ropath"""
first = patch_seq[0]
assert first.difftype != "diff", "First patch in sequence " \
"%s was a diff" % patch_seq
if not first.isreg():
# No need to bother with data if not regular file
assert len(patch_seq) == 1, "Patch sequence isn't regular, but " \
"has %d entries" % len(patch_seq)
return first.get_ropath()
current_file = first.open("rb")
for delta_ropath in patch_seq[1:]:
assert delta_ropath.difftype == "diff", delta_ropath.difftype
if not isinstance(current_file, file):
"""
librsync insists on a real file object, which we create manually
by using the duplicity.tempdir to tell us where.
See https://bugs.launchpad.net/duplicity/+bug/670891 for discussion
of os.tmpfile() vs tempfile.TemporaryFile() w.r.t. Windows / Posix,
which is worked around in librsync.PatchedFile() now.
"""
tempfp = tempfile.TemporaryFile(dir=tempdir.default().dir())
util.copyfileobj(current_file, tempfp)
assert not current_file.close()
tempfp.seek(0)
current_file = tempfp
current_file = librsync.PatchedFile(current_file,
delta_ropath.open("rb"))
result = patch_seq[-1].get_ropath()
result.setfileobj(current_file)
return result
def run_command(cmd, env=None, max_timeout=None):
"""
Run command and return its return status code and its output
"""
arglist = cmd.split()
output = os.tmpfile()
try:
pipe = Popen(arglist, stdout=output, stderr=STDOUT, env=env)
except Exception, errmsg:
return 1, errmsg
# Wait only max_timeout seconds.
if max_timeout:
start = time.time()
while pipe.poll() is None:
time.sleep(0.1)
if time.time() - start > max_timeout:
os.kill(pipe.pid, signal.SIGINT)
pipe.wait()
return 1, "Time exceeded"
pipe.wait()
output.seek(0)
return pipe.returncode, output.read()
def run_command(cmd, env=None, max_timeout=None):
"""
Run command and return its return status code and its output
"""
arglist = cmd.split()
output = os.tmpfile()
try:
pipe = Popen(arglist, stdout=output, stderr=STDOUT, env=env)
except Exception, errmsg:
return 1, errmsg
# Wait only max_timeout seconds.
if max_timeout:
start = time.time()
while pipe.poll() is None:
time.sleep(0.1)
if time.time() - start > max_timeout:
os.kill(pipe.pid, signal.SIGINT)
pipe.wait()
return 1, "Time exceeded"
pipe.wait()
output.seek(0)
return pipe.returncode, output.read()
def test_tmpfile(self):
# As with test_tmpnam() below, the Windows implementation of tmpfile()
# attempts to create a file in the root directory of the current drive.
# On Vista and Server 2008, this test will always fail for normal users
# as writing to the root directory requires elevated privileges. With
# XP and below, the semantics of tmpfile() are the same, but the user
# running the test is more likely to have administrative privileges on
# their account already. If that's the case, then os.tmpfile() should
# work. In order to make this test as useful as possible, rather than
# trying to detect Windows versions or whether or not the user has the
# right permissions, just try and create a file in the root directory
# and see if it raises a 'Permission denied' OSError. If it does, then
# test that a subsequent call to os.tmpfile() raises the same error. If
# it doesn't, assume we're on XP or below and the user running the test
# has administrative privileges, and proceed with the test as normal.
with warnings.catch_warnings():
warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning)
if sys.platform == 'win32':
name = '\\python_test_os_test_tmpfile.txt'
if os.path.exists(name):
os.remove(name)
try:
fp = open(name, 'w')
except IOError, first:
# open() failed, assert tmpfile() fails in the same way.
# Although open() raises an IOError and os.tmpfile() raises an
# OSError(), 'args' will be (13, 'Permission denied') in both
# cases.
try:
fp = os.tmpfile()
except OSError, second:
self.assertEqual(first.args, second.args)
else:
self.fail("expected os.tmpfile() to raise OSError")
return
else:
# open() worked, therefore, tmpfile() should work. Close our
# dummy file and proceed with the test as normal.
fp.close()
os.remove(name)
fp = os.tmpfile()
fp.write("foobar")
fp.seek(0,0)
s = fp.read()
fp.close()
self.assertTrue(s == "foobar")
def test_tmpfile(self):
# As with test_tmpnam() below, the Windows implementation of tmpfile()
# attempts to create a file in the root directory of the current drive.
# On Vista and Server 2008, this test will always fail for normal users
# as writing to the root directory requires elevated privileges. With
# XP and below, the semantics of tmpfile() are the same, but the user
# running the test is more likely to have administrative privileges on
# their account already. If that's the case, then os.tmpfile() should
# work. In order to make this test as useful as possible, rather than
# trying to detect Windows versions or whether or not the user has the
# right permissions, just try and create a file in the root directory
# and see if it raises a 'Permission denied' OSError. If it does, then
# test that a subsequent call to os.tmpfile() raises the same error. If
# it doesn't, assume we're on XP or below and the user running the test
# has administrative privileges, and proceed with the test as normal.
with warnings.catch_warnings():
warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning)
if sys.platform == 'win32':
name = '\\python_test_os_test_tmpfile.txt'
if os.path.exists(name):
os.remove(name)
try:
fp = open(name, 'w')
except IOError, first:
# open() failed, assert tmpfile() fails in the same way.
# Although open() raises an IOError and os.tmpfile() raises an
# OSError(), 'args' will be (13, 'Permission denied') in both
# cases.
try:
fp = os.tmpfile()
except OSError, second:
self.assertEqual(first.args, second.args)
else:
self.fail("expected os.tmpfile() to raise OSError")
return
else:
# open() worked, therefore, tmpfile() should work. Close our
# dummy file and proceed with the test as normal.
fp.close()
os.remove(name)
fp = os.tmpfile()
fp.write("foobar")
fp.seek(0,0)
s = fp.read()
fp.close()
self.assertTrue(s == "foobar")
def test_tmpfile(self):
# As with test_tmpnam() below, the Windows implementation of tmpfile()
# attempts to create a file in the root directory of the current drive.
# On Vista and Server 2008, this test will always fail for normal users
# as writing to the root directory requires elevated privileges. With
# XP and below, the semantics of tmpfile() are the same, but the user
# running the test is more likely to have administrative privileges on
# their account already. If that's the case, then os.tmpfile() should
# work. In order to make this test as useful as possible, rather than
# trying to detect Windows versions or whether or not the user has the
# right permissions, just try and create a file in the root directory
# and see if it raises a 'Permission denied' OSError. If it does, then
# test that a subsequent call to os.tmpfile() raises the same error. If
# it doesn't, assume we're on XP or below and the user running the test
# has administrative privileges, and proceed with the test as normal.
with warnings.catch_warnings():
warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning)
if sys.platform == 'win32':
name = '\\python_test_os_test_tmpfile.txt'
if os.path.exists(name):
os.remove(name)
try:
fp = open(name, 'w')
except IOError, first:
# open() failed, assert tmpfile() fails in the same way.
# Although open() raises an IOError and os.tmpfile() raises an
# OSError(), 'args' will be (13, 'Permission denied') in both
# cases.
try:
fp = os.tmpfile()
except OSError, second:
self.assertEqual(first.args, second.args)
return
else:
if test_support.check_impl_detail(pypy=False):
self.fail("expected os.tmpfile() to raise OSError")
# on PyPy, os.tmpfile() uses the tempfile module
# anyway, so works even if we cannot write in root.
fp.close()
else:
# open() worked, therefore, tmpfile() should work. Close our
# dummy file and proceed with the test as normal.
fp.close()
os.remove(name)
fp = os.tmpfile()
fp.write("foobar")
fp.seek(0,0)
s = fp.read()
fp.close()
self.assertTrue(s == "foobar")
def test_tmpfile(self):
if not hasattr(os, "tmpfile"):
return
# As with test_tmpnam() below, the Windows implementation of tmpfile()
# attempts to create a file in the root directory of the current drive.
# On Vista and Server 2008, this test will always fail for normal users
# as writing to the root directory requires elevated privileges. With
# XP and below, the semantics of tmpfile() are the same, but the user
# running the test is more likely to have administrative privileges on
# their account already. If that's the case, then os.tmpfile() should
# work. In order to make this test as useful as possible, rather than
# trying to detect Windows versions or whether or not the user has the
# right permissions, just try and create a file in the root directory
# and see if it raises a 'Permission denied' OSError. If it does, then
# test that a subsequent call to os.tmpfile() raises the same error. If
# it doesn't, assume we're on XP or below and the user running the test
# has administrative privileges, and proceed with the test as normal.
with warnings.catch_warnings():
warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning)
if sys.platform == 'win32':
name = '\\python_test_os_test_tmpfile.txt'
if os.path.exists(name):
os.remove(name)
try:
fp = open(name, 'w')
except IOError, first:
# open() failed, assert tmpfile() fails in the same way.
# Although open() raises an IOError and os.tmpfile() raises an
# OSError(), 'args' will be (13, 'Permission denied') in both
# cases.
try:
fp = os.tmpfile()
except OSError, second:
self.assertEqual(first.args, second.args)
else:
self.fail("expected os.tmpfile() to raise OSError")
return
else:
# open() worked, therefore, tmpfile() should work. Close our
# dummy file and proceed with the test as normal.
fp.close()
os.remove(name)
fp = os.tmpfile()
fp.write("foobar")
fp.seek(0,0)
s = fp.read()
fp.close()
self.assertTrue(s == "foobar")