本文整理汇总了Python中os.O_TRUNC属性的典型用法代码示例。如果您正苦于以下问题：Python os.O_TRUNC属性的具体用法？Python os.O_TRUNC怎么用？Python os.O_TRUNC使用的例子？那么恭喜您, 这里精选的属性代码示例或许可以为您提供帮助。您也可以进一步了解该属性所在模块os的用法示例。
在下文中一共展示了os.O_TRUNC属性的18个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: store_acs_service_principal
点赞 6
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def store_acs_service_principal(subscription_id, client_secret, service_principal,
file_name=’acsServicePrincipal.json’):
obj = {}
if client_secret:
obj[‘client_secret’] = client_secret
if service_principal:
obj[‘service_principal’] = service_principal
config_path = os.path.join(get_config_dir(), file_name)
full_config = load_service_principals(config_path=config_path)
if not full_config:
full_config = {}
full_config[subscription_id] = obj
with os.fdopen(os.open(config_path, os.O_RDWR | os.O_CREAT | os.O_TRUNC, 0o600),
‘w+’) as spFile:
json.dump(full_config, spFile)
开发者ID:Azure，项目名称:azure-cli-extensions，代码行数:19，
示例2: test_write
点赞 6
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def test_write(self):
test_filename = “tmp.write.test”
# trivial write
fd = os.open(test_filename, flags)
self.assertEqual(os.write(fd, “42”), 2)
os.close(fd)
os.unlink(test_filename)
# write to closed file
fd = os.open(test_filename, flags)
os.close(fd)
self.assertRaisesMessage(OSError, “[Errno 9] Bad file descriptor”, os.write, fd, “42”)
os.unlink(test_filename)
# write to file with wrong permissions
fd = os.open(test_filename, os.O_CREAT | os.O_TRUNC | os.O_RDONLY)
self.assertRaisesMessage(OSError, “[Errno -2146232800] Can not write to ” + test_filename, os.write, fd, “42”)
os.close(fd)
os.unlink(test_filename)
开发者ID:IronLanguages，项目名称:ironpython2，代码行数:22，
示例3: upload_config
点赞 6
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def upload_config(self):
try:
stream = flask.request.stream
file_path = cfg.find_config_files(project=CONF.project,
prog=CONF.prog)[0]
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00600
mode = stat.S_IRUSR | stat.S_IWUSR
with os.fdopen(os.open(file_path, flags, mode), ‘wb’) as cfg_file:
b = stream.read(BUFFER)
while b:
cfg_file.write(b)
b = stream.read(BUFFER)
CONF.mutate_config_files()
except Exception as e:
LOG.error(“Unable to update amphora-agent configuration: “
“{}”.format(str(e)))
return webob.Response(json=dict(
message=”Unable to update amphora-agent configuration.”,
details=str(e)), status=500)
return webob.Response(json={‘message’: ‘OK’}, status=202)
开发者ID:openstack，项目名称:octavia，代码行数:25，
示例4: install_netns_systemd_service
点赞 6
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def install_netns_systemd_service():
os_utils = osutils.BaseOS.get_os_util()
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00644
mode = (stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
# TODO(bcafarel): implement this for other init systems
# netns handling depends on a separate unit file
netns_path = os.path.join(consts.SYSTEMD_DIR,
consts.AMP_NETNS_SVC_PREFIX + ‘.service’)
jinja_env = jinja2.Environment(
autoescape=True, loader=jinja2.FileSystemLoader(os.path.dirname(
os.path.realpath(__file__)
) + consts.AGENT_API_TEMPLATES))
if not os.path.exists(netns_path):
with os.fdopen(os.open(netns_path, flags, mode), ‘w’) as text_file:
text = jinja_env.get_template(
consts.AMP_NETNS_SVC_PREFIX + ‘.systemd.j2’).render(
amphora_nsname=consts.AMPHORA_NAMESPACE,
HasIFUPAll=os_utils.has_ifup_all())
text_file.write(text)
开发者ID:openstack，项目名称:octavia，代码行数:26，
示例5: write_port_interface_file
点赞 6
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def write_port_interface_file(self, netns_interface, fixed_ips, mtu,
interface_file_path, template_port):
# write interface file
# If we are using a consolidated interfaces file, just append
# otherwise clear the per interface file as we are rewriting it
# TODO(johnsom): We need a way to clean out old interfaces records
if CONF.amphora_agent.agent_server_network_file:
flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND
else:
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(interface_file_path, flags, mode),
‘w’) as text_file:
text = self._generate_network_file_text(
netns_interface, fixed_ips, mtu, template_port)
text_file.write(text)
开发者ID:openstack，项目名称:octavia，代码行数:22，
示例6: store_config
点赞 6
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def store_config(self, config, global_only):
“””Persists locally or globally configuration object.
Global configuration is updated only when :global_only: is True,
otherwise, updates are written to local project configuration
“””
filepath = self.global_config_path if global_only else \
self.local_config_path
if global_only:
os.umask(0)
fd = os.open(filepath, os.O_CREAT | os.O_RDWR | os.O_TRUNC, 0o600)
with self.global_config_lock:
with open(fd, ‘w+’) as file:
config.write(file)
else:
with open(filepath, ‘w+’) as file:
config.write(file)
return self.load_config(local_only=True, global_only=True)
开发者ID:SwissDataScienceCenter，项目名称:renku-python，代码行数:22，
示例7: __setitem__
点赞 6
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def __setitem__(self, key, message):
“””Replace the keyed message; raise KeyError if it doesn’t exist.”””
path = os.path.join(self._path, str(key))
try:
f = open(path, ‘rb+’)
except OSError as e:
if e.errno == errno.ENOENT:
raise KeyError(‘No message with key: %s’ % key)
else:
raise
try:
if self._locked:
_lock_file(f)
try:
os.close(os.open(path, os.O_WRONLY | os.O_TRUNC))
self._dump_message(message, f)
if isinstance(message, MHMessage):
self._dump_sequences(message, key)
finally:
if self._locked:
_unlock_file(f)
finally:
_sync_close(f)
开发者ID:Microvellum，项目名称:Fluid-Designer，代码行数:25，
示例8: __init__
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def __init__(self, name, mode):
mode = {
“r”: os.O_RDONLY,
“w”: os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
}[mode]
if hasattr(os, “O_BINARY”):
mode |= os.O_BINARY
self.fd = os.open(name, mode, 0o666)
开发者ID:war-and-code，项目名称:jawfish，代码行数:10，
示例9: create_empty_file
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def create_empty_file(filename):
“””Create an empty file. If the file already exists, truncate it.”””
fd = os.open(filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC)
os.close(fd)
开发者ID:war-and-code，项目名称:jawfish，代码行数:6，
示例10: __init__
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def __init__(self, name, shape, dtype=np.float32, uuid=””):
self.shape = shape
self.dtype = dtype
size = reduce(lambda x, y: x * y, shape)
size_bytes = np.dtype(dtype).itemsize * size
self._mem_path = None
self._mem_pointer = None
if os.name == “nt”:
self._mem_path = “/HOLODECK_MEM” + uuid + “_” + name
self._mem_pointer = mmap.mmap(0, size_bytes, self._mem_path)
elif os.name == “posix”:
self._mem_path = “/dev/shm/HOLODECK_MEM” + uuid + “_” + name
f = os.open(self._mem_path, os.O_CREAT | os.O_TRUNC | os.O_RDWR)
self._mem_file = f
os.ftruncate(f, size_bytes)
os.fsync(f)
# TODO – I think we are leaking a file object here. Unfortunately, we
# can’t just .close() it since numpy acquires a reference to it
# below and I can’t find a way to release it in __linux_unlink__()
self._mem_pointer = mmap.mmap(f, size_bytes)
else:
raise HolodeckException(“Currently unsupported os: ” + os.name)
self.np_array = np.ndarray(shape, dtype=dtype)
self.np_array.data = (Shmem._numpy_to_ctype[dtype] * size).from_buffer(self._mem_pointer)
开发者ID:BYU-PCCL，项目名称:holodeck，代码行数:29，
示例11: install_vmprof
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def install_vmprof(name=”thread”):
cpid = multiprocessing.current_process().name
ctid = threading.current_thread().name
fname = “vmprof-{}-{}-{}-{}.dat”.format(name, cpid, ctid, time.time())
flags = os.O_RDWR | os.O_CREAT | os.O_TRUNC
outfd = os.open(fname, flags)
vmprof.enable(outfd, period=0.01)
# atexit.register(close_profile_file)
开发者ID:fake-name，项目名称:ReadableWebProxy，代码行数:14，
示例12: __init__
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def __init__(self, name, mode):
mode = {
“r”: os.O_RDONLY,
“w”: os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
}[mode]
if hasattr(os, “O_BINARY”):
mode |= os.O_BINARY
self.fd = os.open(name, mode, 0666)
开发者ID:glmcdona，项目名称:meddle，代码行数:10，
示例13: set_sequences
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def set_sequences(self, sequences):
“””Set sequences using the given name-to-key-list dictionary.”””
f = open(os.path.join(self._path, ‘.mh_sequences’), ‘r+’)
try:
os.close(os.open(f.name, os.O_WRONLY | os.O_TRUNC))
for name, keys in sequences.iteritems():
if len(keys) == 0:
continue
f.write(‘%s:’ % name)
prev = None
completing = False
for key in sorted(set(keys)):
if key – 1 == prev:
if not completing:
completing = True
f.write(‘-‘)
elif completing:
completing = False
f.write(‘%s %s’ % (prev, key))
else:
f.write(‘ %s’ % key)
prev = key
if completing:
f.write(str(prev) + ‘\n’)
else:
f.write(‘\n’)
finally:
_sync_close(f)
开发者ID:glmcdona，项目名称:meddle，代码行数:30，
示例14: _acquire
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def _acquire(self):
open_mode = os.O_RDWR | os.O_CREAT | os.O_TRUNC
try:
fd = os.open(self._lock_file, open_mode)
except OSError:
pass
else:
try:
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
except (IOError, OSError):
os.close(fd)
else:
self._lock_file_fd = fd
return None
开发者ID:a4k-openproject，项目名称:a4kScrapers，代码行数:17，
示例15: write_vip_interface_file
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def write_vip_interface_file(self, interface_file_path,
primary_interface, vip, ip, broadcast,
netmask, gateway, mtu, vrrp_ip, vrrp_version,
render_host_routes, template_vip):
# write interface file
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
# If we are using a consolidated interfaces file, just append
# otherwise clear the per interface file as we are rewriting it
# TODO(johnsom): We need a way to clean out old interfaces records
if CONF.amphora_agent.agent_server_network_file:
flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND
else:
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
with os.fdopen(os.open(interface_file_path, flags, mode),
‘w’) as text_file:
text = template_vip.render(
consts=consts,
interface=primary_interface,
vip=vip,
vip_ipv6=ip.version == 6,
# For ipv6 the netmask is already the prefix
prefix=(netmask if ip.version == 6
else utils.netmask_to_prefix(netmask)),
broadcast=broadcast,
netmask=netmask,
gateway=gateway,
network=utils.ip_netmask_to_cidr(vip, netmask),
mtu=mtu,
vrrp_ip=vrrp_ip,
vrrp_ipv6=vrrp_version == 6,
host_routes=render_host_routes,
topology=CONF.controller_worker.loadbalancer_topology,
)
text_file.write(text)
开发者ID:openstack，项目名称:octavia，代码行数:39，
示例16: write_interfaces_file
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def write_interfaces_file(self):
name = ‘/etc/netns/{}/network/interfaces’.format(
consts.AMPHORA_NAMESPACE)
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(name, flags, mode), ‘w’) as int_file:
int_file.write(‘auto lo\n’)
int_file.write(‘iface lo inet loopback\n’)
if not CONF.amphora_agent.agent_server_network_file:
int_file.write(‘source /etc/netns/{}/network/’
‘interfaces.d/*.cfg\n’.format(
consts.AMPHORA_NAMESPACE))
开发者ID:openstack，项目名称:octavia，代码行数:15，
示例17: write_port_interface_if_local_scripts
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def write_port_interface_if_local_scripts(
self, template_script, ifup=True):
file_name = ‘ifup’ + ‘-local’
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
if not ifup:
file_name = ‘ifdown’ + ‘-local’
with os.fdopen(
os.open(os.path.join(
‘/sbin/’, file_name), flags, mode), ‘w’) as text_file:
text = template_script.render()
text_file.write(text)
os.chmod(os.path.join(‘/sbin/’, file_name), stat.S_IEXEC)
开发者ID:openstack，项目名称:octavia，代码行数:15，
示例18: upload_server_cert
点赞 5
# 需要导入模块: import os [as 别名]
# 或者: from os import O_TRUNC [as 别名]
def upload_server_cert():
stream = flask.request.stream
file_path = CONF.amphora_agent.agent_server_cert
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00600
mode = stat.S_IRUSR | stat.S_IWUSR
with os.fdopen(os.open(file_path, flags, mode), ‘wb’) as crt_file:
b = stream.read(BUFFER)
while b:
crt_file.write(b)
b = stream.read(BUFFER)
return webob.Response(json={‘message’: ‘OK’}, status=202)
开发者ID:openstack，项目名称:octavia，代码行数:15，
注：本文中的os.O_TRUNC属性示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。
今天的文章python trunc_Python os.O_TRUNC属性代码示例分享到此就结束了，感谢您的阅读。
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
如需转载请保留出处：https://bianchenghao.cn/9572.html
