Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _log_default(self):
"""wrap loggers for this application."""
return wrap_logger_handlers(LocalProcessSpawner._log_default(self))
# TERM failed, use KILL
status = await self.poll()
if status is not None:
return
self.log.debug("Killing %i", self.pid)
await self._signal(signal.SIGKILL)
await self.wait_for_death(self.kill_timeout)
status = await self.poll()
if status is None:
# it all failed, zombie process
self.log.warning("Process %i never died", self.pid)
class SimpleLocalProcessSpawner(LocalProcessSpawner):
"""
A version of LocalProcessSpawner that doesn't require users to exist on
the system beforehand.
Only use this for testing.
Note: DO NOT USE THIS FOR PRODUCTION USE CASES! It is very insecure, and
provides absolutely no isolation between different users!
"""
home_dir_template = Unicode(
'/tmp/{username}',
config=True,
help="""
Template to expand to set the user home.
{username} is expanded to the jupyterhub username.
return self.child_spawner.poll()
else:
return _yield_val(1)
class ProfilesSpawner(WrapSpawner):
"""ProfilesSpawner - leverages the Spawner options form feature to allow user-driven
configuration of Spawner classes while permitting:
1) configuration of Spawner classes that don't natively implement options_form
2) administrator control of allowed configuration changes
3) runtime choice of which Spawner backend to launch
"""
profiles = List(
trait = Tuple( Unicode(), Unicode(), Type(Spawner), Dict() ),
default_value = [ ( 'Local Notebook Server', 'local', LocalProcessSpawner,
{'start_timeout': 15, 'http_timeout': 10} ) ],
minlen = 1,
config = True,
help = """List of profiles to offer for selection. Signature is:
List(Tuple( Unicode, Unicode, Type(Spawner), Dict )) corresponding to
profile display name, unique key, Spawner class, dictionary of spawner config options.
The first three values will be exposed in the input_template as {display}, {key}, and {type}"""
)
child_profile = Unicode()
form_template = Unicode(
"""<label for="profile">Select a job profile:</label>
<select required="" name="profile" class="form-control">
{input_template}</select>
import os
import pipes
import shutil
from subprocess import Popen
from jupyterhub.spawner import LocalProcessSpawner
from jupyterhub.utils import random_port
class CylcSpawner01(LocalProcessSpawner):
def __init__(self, *args, **kwargs):
super(CylcSpawner01, self).__init__(*args, **kwargs)
async def start(self):
"""Start the single-user server."""
self.port = random_port()
cmd = []
env = self.get_env()
cmd.extend([
"python",
os.path.join(os.path.dirname(os.path.realpath(__file__)), "cylc-singleuser.py"),
str(self.port)
])
# cmd.extend(self.get_args())
@property
def spawner_class(self):
return self.settings.get('spawner_class', LocalProcessSpawner)
c.JupyterHub.bind_url = 'http://:9002'
c.Authenticator.add_user_cmd = ['adduser', '--home', '/home/USERNAME']
c.LocalAuthenticator.create_system_users = True
c.Authenticator.delete_invalid_users = True
c.JupyterHub.authenticator_class = 'sqliteauthenticator.SQLiteAuthenticator'
c.Authenticator.admin_users = {'admin'}
from jupyterhub.spawner import LocalProcessSpawner
class MySpawner(LocalProcessSpawner):
def _notebook_dir_default(self):
return '/home/' + self.user.name
c.JupyterHub.spawner_class = MySpawner
spawn the server on 10.0.1.2:443 with https:
jupyterhub --ip 10.0.1.2 --port 443 --ssl-key my_ssl.key --ssl-cert my_ssl.cert
"""
aliases = Dict(aliases)
flags = Dict(flags)
subcommands = {
'token': (NewToken, "Generate an API token for a user"),
'upgrade-db': (UpgradeDB, "Upgrade your JupyterHub state database to the current version."),
}
classes = List([
Spawner,
LocalProcessSpawner,
Authenticator,
PAMAuthenticator,
])
load_groups = Dict(List(Unicode()),
help="""Dict of 'group': ['usernames'] to load at startup.
This strictly *adds* groups and users to groups.
Loading one set of groups, then starting JupyterHub again with a different
set will not remove users or groups from previous launches.
That must be done through the API.
"""
).tag(config=True)
config_file = Unicode('jupyterhub_config.py',
import docker
except ImportError:
pass
# Utility to create dummy Futures to return values through yields
def _yield_val(x=None):
f = concurrent.Future()
f.set_result(x)
return f
class WrapSpawner(Spawner):
# Grab this from constructor args in case some Spawner ever wants it
config = Any()
child_class = Type(LocalProcessSpawner, Spawner,
config=True,
help="""The class to wrap for spawning single-user servers.
Should be a subclass of Spawner.
"""
)
child_config = Dict(default_value={},
config=True,
help="Dictionary of config values to apply to wrapped spawner class."
)
child_state = Dict(default_value={})
child_spawner = Instance(Spawner, allow_none=True)
def construct_child(self):
def select_profile(self, profile):
self.log.debug("select_profile: " + profile)
if profile == "":
self.child_class, self.child_config = LocalProcessSpawner, {}
else:
try:
self.child_class, self.child_config = self.spawners[profile]
except KeyError:
raise web.HTTPError(404)
def spawner_class(self):
return self.settings.get('spawner_class', LocalProcessSpawner)