How to use the ipykernel.connect function in ipykernel

To help you get started, we’ve selected a few ipykernel 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 microsoft / pai / contrib / python-sdk / openpaisdk / notebook.py View on Github external
def get_notebook_path():
    """
    Return the full path of the jupyter notebook.
    Reference: https://github.com/jupyter/notebook/issues/1000#issuecomment-359875246
    """
    import requests
    from requests.compat import urljoin
    from notebook.notebookapp import list_running_servers
    import ipykernel

    kernel_id = re.search('kernel-(.*).json',
                          ipykernel.connect.get_connection_file()).group(1)
    servers = list_running_servers()
    for ss in servers:
        response = requests.get(urljoin(ss['url'], 'api/sessions'),
                                params={'token': ss.get('token', '')})
        info = json.loads(response.text)
        if isinstance(info, dict) and info['message'] == 'Forbidden':
            continue
        for nn in info:
            if nn['kernel']['id'] == kernel_id:
                relative_path = nn['notebook']['path']
                return os.path.join(ss['notebook_dir'], relative_path)
github JovianML / jovian-py / jovian / utils / jupyter.py View on Github external
def get_notebook_path_py():
    try:  # Python 3
        from notebook.notebookapp import list_running_servers
    except ImportError:  # no-cover
        # Python 2
        import warnings
        from IPython.utils.shimmodule import ShimWarning
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", category=ShimWarning)
            from IPython.html.notebookapp import list_running_servers

    import ipykernel
    """Get the full path of the jupyter notebook."""
    kernel_id = re.search('kernel-(.*).json',
                          ipykernel.connect.get_connection_file()).group(1)
    servers = list_running_servers()
    for ss in servers:
        response = requests.get(urljoin(ss['url'], 'api/sessions'),
                                params={'token': ss.get('token', '')})
        if response.status_code == 200:
            for nn in json.loads(response.text):
                if nn["kernel"] and nn['kernel']['id'] == kernel_id:
                    relative_path = nn['notebook']['path']
                    return os.path.join(ss['notebook_dir'], relative_path)
github jwkvam / bowtie / bowtie / _magic.py View on Github external
def get_notebook_name() -> str:
    """Return the full path of the jupyter notebook.

    References
    ----------
    https://github.com/jupyter/notebook/issues/1000#issuecomment-359875246

    """
    kernel_id = re.search(  # type: ignore
        'kernel-(.*).json', ipykernel.connect.get_connection_file()
    ).group(1)
    servers = list_running_servers()
    for server in servers:
        response = requests.get(
            urljoin(server['url'], 'api/sessions'), params={'token': server.get('token', '')}
        )
        for session in json.loads(response.text):
            if session['kernel']['id'] == kernel_id:
                relative_path = session['notebook']['path']
                return pjoin(server['notebook_dir'], relative_path)
    raise Exception('Noteboook not found.')
github JovianML / jovian-py / jovian / utils / jupyter.py View on Github external
def get_notebook_server_path():
    try:  # Python 3
        from notebook.notebookapp import list_running_servers
    except ImportError:  # no-cover
        # Python 2
        import warnings
        from IPython.utils.shimmodule import ShimWarning
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", category=ShimWarning)
            from IPython.html.notebookapp import list_running_servers

    """Get the path of the notebook relative to the Jupyter server"""
    import ipykernel
    kernel_id = re.search('kernel-(.*).json',
                          ipykernel.connect.get_connection_file()).group(1)
    servers = list_running_servers()
    for ss in servers:
        response = requests.get(urljoin(ss['url'], 'api/sessions'),
                                params={'token': ss.get('token', '')})
        for nn in json.loads(response.text):
            if nn['kernel']['id'] == kernel_id:
                relative_path = nn['notebook']['path']
                return relative_path
github WorldWideTelescope / pywwt / pywwt / jupyter_server.py View on Github external
Copied from
    https://github.com/jupyter/notebook/issues/3156#issuecomment-401119433
    with miniscule changes. This is gross, but appears to be the best
    available option right now.

    """
    import ipykernel
    import json
    from notebook.notebookapp import list_running_servers
    import re
    import requests

    # First, find our ID.
    kernel_id = re.search(
        'kernel-(.*).json',
        ipykernel.connect.get_connection_file()
    ).group(1)

    # Now, check all of the running servers known on this machine. We have to
    # talk to each server to figure out if it's ours or somebody else's.
    for s in list_running_servers():
        response = requests.get(
            requests.compat.urljoin(s['url'], 'api/sessions'),
            params={'token': s.get('token', '')}
        )

        for n in json.loads(response.text):
            if n['kernel']['id'] == kernel_id:
                return s['base_url']  # Found it!

    raise Exception('cannot locate our notebook server; is this code running in a Jupyter kernel?')
github cknoll / ipydex / ipydex / core.py View on Github external
try:
        import requests
        from requests.compat import urljoin
    except ImportError:
        msg = "This functions depends on the module requests."
        # it is not an official dependency because this is not a core functionality
        raise ImportError(msg)

    import ipykernel
    import json
    import re
    from notebook.notebookapp import list_running_servers

    kernel_id = re.search('kernel-(.*).json',
                          ipykernel.connect.get_connection_file()).group(1)
    servers = list_running_servers()
    for ss in servers:
        response = requests.get(urljoin(ss['url'], 'api/sessions'),
                                params={'token': ss.get('token', '')})
        for nn in json.loads(response.text):
            if nn['kernel']['id'] == kernel_id:
                relative_path = nn['notebook']['path']
                return os.path.join(ss['notebook_dir'], relative_path)
github wandb / client / wandb / jupyter.py View on Github external
def notebook_metadata():
    """Attempts to query jupyter for the path and name of the notebook file"""
    error_message = "Failed to query for notebook name, you can set it manually with the WANDB_NOTEBOOK_NAME environment variable"
    try:
        import ipykernel
        from notebook.notebookapp import list_running_servers
        kernel_id = re.search('kernel-(.*).json', ipykernel.connect.get_connection_file()).group(1)
        servers = list(list_running_servers())  # TODO: sometimes there are invalid JSON files and this blows up
    except Exception:
        logger.error(error_message)
        return {}
    for s in servers:
        try:
            if s['password']:
                raise ValueError("Can't query password protected kernel")
            res = requests.get(urljoin(s['url'], 'api/sessions'), params={'token': s.get('token', '')}).json()
        except (requests.RequestException, ValueError):
            logger.error(error_message)
            return {}
        for nn in res:
            # TODO: wandb/client#400 found a case where res returned an array of strings...
            if isinstance(nn, dict) and nn.get("kernel"):
                if nn['kernel']['id'] == kernel_id:
github enthought / envisage / envisage / plugins / ipython_kernel / internal_ipkernel.py View on Github external
def new_qt_console(self):
        """ Start a new qtconsole connected to our kernel. """
        console = ipykernel.connect.connect_qtconsole(
            self.ipkernel.connection_file, argv=["--no-confirm-exit"],
        )
        self.consoles.append(console)
        return console
github nuclio / nuclio-jupyter / nuclio / utils.py View on Github external
def notebook_file_name(ikernel):
    """Return the full path of the jupyter notebook."""

    # the following code won't work when the notebook is being executed
    # through running `jupyter nbconvert --execute` this env var enables to
    # overcome it
    file_name = environ.get('JUPYTER_NOTEBOOK_FILE_NAME')
    if file_name is not None:
        return file_name

    # Check that we're running under notebook
    if not (ikernel and ikernel.config['IPKernelApp']):
        return

    kernel_id = re.search('kernel-(.*).json',
                          ipykernel.connect.get_connection_file()).group(1)
    servers = list_running_servers()
    for srv in servers:
        query = {'token': srv.get('token', '')}
        url = urljoin(srv['url'], 'api/sessions') + '?' + urlencode(query)
        for session in json.load(urlopen(url)):
            if session['kernel']['id'] == kernel_id:
                relative_path = session['notebook']['path']
                return path.join(srv['notebook_dir'], relative_path)
github wbuchwalter / fairing / fairing / notebook_helper.py View on Github external
def get_notebook_name():
    """
    Return the full path of the jupyter notebook.
    """
    kernel_id = re.search('kernel-(.*).json',
                          ipykernel.connect.get_connection_file()).group(1)
    servers = list_running_servers()
    for ss in servers:
        response = requests.get(urljoin(ss['url'], 'api/sessions'),
                                params={'token': ss.get('token', '')})
        for nn in json.loads(response.text):
            if nn['kernel']['id'] == kernel_id:
                full_path = nn['notebook']['path']
                return os.path.basename(full_path)
    
    return f