Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def init_proxy(self):
"""Load the Proxy config into the database"""
self.proxy = self.db.query(orm.Proxy).first()
if self.proxy is None:
self.proxy = orm.Proxy(
public_server=orm.Server(),
api_server=orm.Server(),
)
self.db.add(self.proxy)
self.db.commit()
self.proxy.auth_token = self.proxy_auth_token # not persisted
self.proxy.log = self.log
self.proxy.public_server.ip = self.ip
self.proxy.public_server.port = self.port
self.proxy.public_server.base_url = self.base_url
self.proxy.api_server.ip = self.proxy_api_ip
self.proxy.api_server.port = self.proxy_api_port
self.proxy.api_server.base_url = '/api/routes/'
self.db.commit()
if False:
JupyterHub expects only one single-server per user
url of the server will be /user/:name
if True:
JupyterHub expects more than one single-server per user
url of the server will be /user/:name/:server_name
"""
db = self.db
if handler:
await self.refresh_auth(handler)
base_url = url_path_join(self.base_url, server_name) + '/'
orm_server = orm.Server(base_url=base_url)
db.add(orm_server)
note = "Server at %s" % base_url
api_token = self.new_api_token(note=note)
db.commit()
spawner = self.spawners[server_name]
spawner.server = server = Server(orm_server=orm_server)
assert spawner.orm_spawner.server is orm_server
# pass requesting handler to the spawner
# e.g. for processing GET params
spawner.handler = handler
# Passing user_options to the spawner
if options is None:
# options unspecified, load from db which should have the previous value
def set_service_cookie(self, user):
"""set the login cookie for services"""
self._set_user_cookie(user, orm.Server(
cookie_name='jupyterhub-services',
base_url=url_path_join(self.base_url, 'services')
))
)
else:
# ensure provided token is registered
self.service_tokens[service.api_token] = service.name
else:
self.service_tokens[service.api_token] = service.name
if service.url:
parsed = urlparse(service.url)
if parsed.port is not None:
port = parsed.port
elif parsed.scheme == 'http':
port = 80
elif parsed.scheme == 'https':
port = 443
server = service.orm.server = orm.Server(
proto=parsed.scheme,
ip=parsed.hostname,
port=port,
cookie_name='jupyterhub-services',
base_url=service.prefix,
)
self.db.add(server)
else:
service.orm.server = None
if service.oauth_available:
self.oauth_provider.add_client(
client_id=service.oauth_client_id,
client_secret=service.api_token,
redirect_uri=service.oauth_redirect_uri,
@classmethod
def find(cls, db, name):
"""Find a user by name.
Returns None if not found.
"""
return db.query(cls).filter(cls.name == name).first()
class Spawner(Base):
""""State about a Spawner"""
__tablename__ = 'spawners'
id = Column(Integer, primary_key=True, autoincrement=True)
user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'))
server_id = Column(Integer, ForeignKey('servers.id', ondelete='SET NULL'))
server = relationship(Server, cascade="all")
state = Column(JSONDict)
name = Column(Unicode(255))
started = Column(DateTime)
last_activity = Column(DateTime, nullable=True)
class Service(Base):
"""A service run with JupyterHub
A service is similar to a User without a Spawner.
A service can have API tokens for accessing the Hub's API
It has:
- name
def set_service_cookie(self, user):
"""set the login cookie for services"""
self._set_user_cookie(user, orm.Server(
cookie_name='jupyterhub-services',
base_url=url_path_join(self.base_url, 'services')
))