Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
Print the results of the parse
:param res: Results of the parsing
:param tzinfo: Timezone found in logs
:param app: Name of application for the logs
"""
mins_of_data = len(res)
total_reqs = sum([x["num_uris"] for x in res])
max_reqs = max([x["num_uris"] for x in res])
min_reqs = min([x["num_uris"] for x in res])
avg_reqs = total_reqs // mins_of_data
first, last = res[0], res[-1]
first, last = first["timestamp"], last["timestamp"]
first = pendulum.from_timestamp(first, tzinfo)
last = pendulum.from_timestamp(last, tzinfo)
first, last = first.isoformat()[:16], last.isoformat()[:16]
test_params = {
"base_url": "http://$your_base_url",
"rate": 100,
"replay_start_time": first,
"replay_end_time": last,
"identifier": "oss",
}
dump = json.dumps(test_params, indent=2)
click.echo(f"{mins_of_data} minutes of traffic data was uploaded to S3.")
click.echo(f"Average requests/min: {avg_reqs}")
click.echo(f"Max requests/min: {max_reqs}")
click.echo(f"Min requests/min: {min_reqs}")
click.echo(f"Timezone found in logs: {tzinfo.name}")
def test_create_from_timestamp_returns_pendulum():
d = pendulum.from_timestamp(pendulum.datetime(1975, 5, 21, 22, 32, 5).timestamp())
assert_datetime(d, 1975, 5, 21, 22, 32, 5)
assert d.timezone_name == "UTC"
def _dt_from_gm_timestamp(gm_timestamp):
return pendulum.from_timestamp(gm_utils.from_gm_timestamp(gm_timestamp))
async def add_headers_and_log(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time_ms = round((time.time() - start_time) * 1000, 2)
response.headers["X-process-time-ms"] = str(process_time_ms)
request_data = {
"method": request.method,
"path": request.url.path,
"request_time": pendulum.from_timestamp(start_time).to_datetime_string(),
"client": {"host": request.client.host, "port": request.client.port},
"elapsed_ms": str(process_time_ms),
"status_code": response.status_code,
}
print(request_data)
return response
def user_last_use_time(self):
t = pendulum.from_timestamp(self.last_use_time, tz=settings.TIME_ZONE)
return t
def as_datetime(self, value):
"""
Return a timestamp as a datetime.
:rtype: pendulum.Pendulum
"""
if isinstance(value, basestring):
return pendulum.parse(value)
if isinstance(value, (int, float)):
return pendulum.from_timestamp(value)
return pendulum.instance(value)
def getCache(url: str) -> Union[dict, bool]:
hashed = hashlib.sha224(str(url).encode('utf-8')).hexdigest()
cache = os.path.normpath(jano_index() + os.path.normcase("/cache/{0}.jcache".format(hashed)))
if os.path.isfile(cache):
time = pendulum.from_timestamp(os.path.getmtime(cache))
now = pendulum.now(Config.values()['timezone'])
if now.diff(time).in_days() >= 2:
os.remove(cache)
return False
return pickle.load(open(cache, "rb"))
else:
return False