Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
spawner = user.orm_spawners[server_name]
if (not spawner.last_activity) or last_activity > spawner.last_activity:
self.log.debug(
"Activity on server %s/%s: %s",
user.name,
server_name,
isoformat(last_activity),
)
spawner.last_activity = last_activity
else:
self.log.debug(
"Not updating server activity on %s/%s: %s < %s",
user.name,
server_name,
isoformat(last_activity),
isoformat(user.last_activity),
)
self.db.commit()
if (not spawner.last_activity) or last_activity > spawner.last_activity:
self.log.debug(
"Activity on server %s/%s: %s",
user.name,
server_name,
isoformat(last_activity),
)
spawner.last_activity = last_activity
else:
self.log.debug(
"Not updating server activity on %s/%s: %s < %s",
user.name,
server_name,
isoformat(last_activity),
isoformat(user.last_activity),
)
self.db.commit()
def server_model(self, spawner, include_state=False):
"""Get the JSON model for a Spawner"""
return {
'name': spawner.name,
'last_activity': isoformat(spawner.orm_spawner.last_activity),
'started': isoformat(spawner.orm_spawner.started),
'pending': spawner.pending,
'ready': spawner.ready,
'state': spawner.get_state() if include_state else None,
'url': url_path_join(spawner.user.url, spawner.name, '/'),
'user_options': spawner.user_options,
'progress_url': spawner._progress_url,
}
(we use naïve utc timestamps everywhere)
"""
try:
dt = parse_date(timestamp)
except Exception:
raise web.HTTPError(400, "Not a valid timestamp: %r", timestamp)
if dt.tzinfo:
# strip timezone info to naïve UTC datetime
dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
now = datetime.utcnow()
if (dt - now) > timedelta(minutes=59):
raise web.HTTPError(
400,
"Rejecting activity from more than an hour in the future: {}".format(
isoformat(dt)
),
)
return dt
if token.user:
owner_key = 'user'
owner = token.user.name
else:
owner_key = 'service'
owner = token.service.name
model = {
owner_key: owner,
'id': token.api_id,
'kind': kind,
'created': isoformat(token.created),
'last_activity': isoformat(token.last_activity),
'expires_at': isoformat(expires_at),
}
model.update(extra)
return model
)
if token.user:
owner_key = 'user'
owner = token.user.name
else:
owner_key = 'service'
owner = token.service.name
model = {
owner_key: owner,
'id': token.api_id,
'kind': kind,
'created': isoformat(token.created),
'last_activity': isoformat(token.last_activity),
'expires_at': isoformat(expires_at),
}
model.update(extra)
return model
def user_model(self, user, include_servers=False, include_state=False):
"""Get the JSON model for a User object"""
if isinstance(user, orm.User):
user = self.users[user.id]
model = {
'kind': 'user',
'name': user.name,
'admin': user.admin,
'groups': [g.name for g in user.groups],
'server': user.url if user.running else None,
'pending': None,
'created': isoformat(user.created),
'last_activity': isoformat(user.last_activity),
}
if '' in user.spawners:
model['pending'] = user.spawners[''].pending
if not include_servers:
model['servers'] = None
return model
servers = model['servers'] = {}
for name, spawner in user.spawners.items():
# include 'active' servers, not just ready
# (this includes pending events)
if spawner.active:
servers[name] = self.server_model(spawner, include_state=include_state)
return model