Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def main(self, top_level=None, base_path=None):
"""
Check command line arguments and run the build steps.
:param top_level: The path to the root of the checked out flocker
directory.
:param base_path: ignored.
"""
to_file(self.sys_module.stderr)
distributions = available_distributions(top_level)
options = BuildOptions(distributions)
try:
options.parseOptions(self.sys_module.argv[1:])
except usage.UsageError as e:
self.sys_module.stderr.write("%s\n" % (options,))
self.sys_module.stderr.write("%s\n" % (e,))
raise SystemExit(1)
self.build_command(
destination_path=options['destination-path'],
package_uri=options['package-uri'],
top_level=top_level,
distribution=options['distribution'],
def _start_logging():
# Name log file based on PID, so different processes so stomp on each
# others' logfiles:
to_file(open("{}.log".format(getpid()), "a"))
"""
Cross-process log tracing: HTTP client.
"""
from __future__ import unicode_literals
import sys
import requests
from eliot import to_file, start_action, add_global_fields
add_global_fields(process="client")
to_file(sys.stdout)
def remote_divide(x, y):
with start_action(action_type="http_request", x=x, y=y) as action:
task_id = action.serialize_task_id()
response = requests.get(
"http://localhost:5000/?x={}&y={}".format(x, y),
headers={"x-eliot-task-id": task_id})
response.raise_for_status() # ensure this is a successful response
result = float(response.text)
action.add_success_fields(result=result)
return result
if __name__ == '__main__':
with start_action(action_type="main"):
from sys import stdout
from eliot import Message, to_file
to_file(stdout)
class Place(object):
def __init__(self, name, contained=()):
self.name = name
self.contained = contained
def visited(self, people):
Message.log(message_type="visited",
people=people, place=self.name)
for thing in self.contained:
thing.visited(people)
def honeymoon(family, destination):
Message.log(message_type="honeymoon", people=family)
from eliot import start_action, to_file
import trio
to_file(open("trio.log", "w"))
async def say(message, delay):
with start_action(action_type="say", message=message):
await trio.sleep(delay)
async def main():
with start_action(action_type="main"):
async with trio.open_nursery() as nursery:
nursery.start_soon(say, "hello", 1)
nursery.start_soon(say, "world", 2)
trio.run(main)
def _setup_eliot():
# type: () -> bool
"""
Set up the `eliot` package so that it outputs to a log file.
"""
global _eliot_configured
if not _eliot_configured:
eliot.to_file(LOG_FILE)
_eliot_configured = True
return _eliot_configured
"""
Output a few Eliot message to standard out.
"""
from __future__ import unicode_literals
import sys
import time
from eliot import log_message, to_file
to_file(sys.stdout)
def main():
log_message(message_type="test", value="hello", another=1)
time.sleep(0.2)
log_message(message_type="test", value="goodbye", another=2)
if __name__ == '__main__':
main()
from sys import stdout
from eliot import start_action, to_file
to_file(stdout)
class Place(object):
def __init__(self, name, contained=()):
self.name = name
self.contained = contained
def visited(self, people):
# No need to repetitively log people, since caller will:
with start_action(action_type="visited", place=self.name):
for thing in self.contained:
thing.visited(people)
def honeymoon(family, destination):
with start_action(action_type="honeymoon", people=family):