How to use the pendulum.from_timestamp function in pendulum

To help you get started, we’ve selected a few pendulum examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github edmunds / shadowreader / shadowreader / parser.py View on Github external
"""
    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}")
github sdispater / pendulum / tests / datetime / test_create_from_timestamp.py View on Github external
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"
github thebigmunch / google-music-scripts / src / google_music_scripts / core.py View on Github external
def _dt_from_gm_timestamp(gm_timestamp):
		return pendulum.from_timestamp(gm_utils.from_gm_timestamp(gm_timestamp))
github Gingernaut / microAuth / app / create_app.py View on Github external
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
github Ehco1996 / django-sspanel / apps / ssserver / models.py View on Github external
def user_last_use_time(self):
        t = pendulum.from_timestamp(self.last_use_time, tz=settings.TIME_ZONE)
        return t
github sdispater / orator / orator / orm / model.py View on Github external
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)
github verifiqueme / core / jano / controllers / CacheController.py View on Github external
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