jupyterhub_config.py

c = get_config()  # noqa

# create a service with permission to start servers
# (and a silly token to test with)
c.JupyterHub.load_roles = [
    {
        "name": "spawn-test",
        "services": ["spawn-test"],
        "scopes": ["read:users", "servers"],
    },
]
c.JupyterHub.services = [
    {
        "name": "spawn-test",
        "api_token": "test-spawn-test",
    },
]

# allow a single user to be launched
c.Authenticator.allowed_users = {"test-user"}
c.JupyterHub.authenticator_class = "null"

# define test spawner that produces progress and takes a long time to start
import asyncio
import time

from jupyterhub.spawner import SimpleLocalProcessSpawner

class ProgressTestSpawner(SimpleLocalProcessSpawner):
    async def progress(self):
        tic = time.perf_counter()
        # this will go past 100% if the progress API waits after start finishes
        for step in range(10, 200, 10):
            elapsed = time.perf_counter() - tic
            yield {"progress": step, "message": f"{elapsed:.1f}s..."}
            await asyncio.sleep(1)

    async def start(self):
        # begin starting the server
        f = asyncio.ensure_future(super().start())
        # make sure it takes at least 10 seconds
        await asyncio.sleep(10)
        # return original result (and keep waiting, in case it still isn't done)
        return await f


c.JupyterHub.spawner_class = ProgressTestSpawner

start-server.py

#!/usr/bin/env python3
"""
Example of starting/stopping a server via the JupyterHub API

1. get user status
2. start server
3. wait for server to be ready via progress api
4. stop server via API
5. wait for server to finish stopping
"""
import json
import logging

import requests

log = logging.getLogger(__name__)

token = "test-spawn-test"


def event_stream(session, url):
    """Generator yielding events from a JSON event stream

    For use with the server progress API
    """
    r = session.get(url, stream=True)
    r.raise_for_status()
    for line in r.iter_lines():
        line = line.decode('utf8', 'replace')
        # event lines all start with `data:`
        # all other lines should be ignored (they will be empty)
        if line.startswith('data:'):
            yield json.loads(line.split(':', 1)[1])


def start_server(session, hub_url, user, server_name=""):
    """Start a server for a jupyterhub user

    Returns the full URL for accessing the server
    """
    user_url = f"{hub_url}/hub/api/users/{user}"
    log_name = f"{user}/{server_name}".rstrip("/")

    # step 1: get user status
    r = session.get(user_url)
    r.raise_for_status()
    user_model = r.json()

    # if server is not 'active', request launch
    if server_name not in user_model.get('servers', {}):
        log.info(f"Starting server {log_name}")
        try:
            r = session.post(f"{user_url}/servers/{server_name}", timeout=0.1)
        except requests.ReadTimeout:
            # don't wait for response, proceed directly to progress API
            pass
        r = session.get(user_url)
        r.raise_for_status()
        user_model = r.json()

    # report server status
    server = user_model['servers'][server_name]
    if server['pending']:
        status = f"pending {server['pending']}"
    elif server['ready']:
        status = "ready"
    else:
        # shouldn't be possible!
        raise ValueError(f"Unexpected server state: {server}")

    log.info(f"Server {log_name} is {status}")

    # wait for server to be ready using progress API
    progress_url = user_model['servers'][server_name]['progress_url']
    for event in event_stream(session, f"{hub_url}{progress_url}"):
        log.info(f"Progress {event['progress']}%: {event['message']}")
        if event.get("ready"):
            server_url = event['url']
            break
    else:
        # server never ready
        raise ValueError(f"{log_name} never started!")

    # at this point, we know the server is ready and waiting to receive requests
    # return the full URL where the server can be accessed
    return f"{hub_url}{server_url}"


def stop_server(session, hub_url, user, server_name=""):
    """Stop a server via the JupyterHub API

    Returns when the server has finished stopping
    """
    # step 1: get user status
    user_url = f"{hub_url}/hub/api/users/{user}"
    server_url = f"{user_url}/servers/{server_name}"
    log_name = f"{user}/{server_name}".rstrip("/")

    log.info(f"Stopping server {log_name}")
    r = session.delete(server_url)
    if r.status_code == 404:
        log.info(f"Server {log_name} already stopped")

    r.raise_for_status()
    if r.status_code == 204:
        log.info(f"Server {log_name} stopped")
        return

    # else: 202, stop requested, but not complete
    log.info(f"Server {log_name} stopping...")


def main():
    """Start and stop one server

    Uses test-user and hub from jupyterhub_config.py in this directory
    """
    user = "test-user"
    hub_url = "http://127.0.0.1:8000"
    session = requests.Session()
    session.headers = {"Authorization": f"token {token}"}
    start_server(session, hub_url, user)
    stop_server(session, hub_url, user)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
    main()