/usr/lib/python2.7/dist-packages/IPython/html/services/sessions/sessionmanager.py is in ipython-notebook 2.3.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 | """A base class session manager.
Authors:
* Zach Sailer
"""
#-----------------------------------------------------------------------------
# Copyright (C) 2013 The IPython Development Team
#
# Distributed under the terms of the BSD License. The full license is in
# the file COPYING, distributed as part of this software.
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
import uuid
import sqlite3
from tornado import web
from IPython.config.configurable import LoggingConfigurable
from IPython.utils.py3compat import unicode_type
#-----------------------------------------------------------------------------
# Classes
#-----------------------------------------------------------------------------
class SessionManager(LoggingConfigurable):
# Session database initialized below
_cursor = None
_connection = None
_columns = {'session_id', 'name', 'path', 'kernel_id'}
@property
def cursor(self):
"""Start a cursor and create a database called 'session'"""
if self._cursor is None:
self._cursor = self.connection.cursor()
self._cursor.execute("""CREATE TABLE session
(session_id, name, path, kernel_id)""")
return self._cursor
@property
def connection(self):
"""Start a database connection"""
if self._connection is None:
self._connection = sqlite3.connect(':memory:')
self._connection.row_factory = self.row_factory
return self._connection
def __del__(self):
"""Close connection once SessionManager closes"""
self.cursor.close()
def session_exists(self, name, path):
"""Check to see if the session for a given notebook exists"""
self.cursor.execute("SELECT * FROM session WHERE name=? AND path=?", (name, path))
reply = self.cursor.fetchone()
if reply is None:
return False
else:
return True
def new_session_id(self):
"Create a uuid for a new session"
return unicode_type(uuid.uuid4())
def create_session(self, name=None, path=None, kernel_id=None):
"""Creates a session and returns its model"""
session_id = self.new_session_id()
return self.save_session(session_id, name=name, path=path, kernel_id=kernel_id)
def save_session(self, session_id, name=None, path=None, kernel_id=None):
"""Saves the items for the session with the given session_id
Given a session_id (and any other of the arguments), this method
creates a row in the sqlite session database that holds the information
for a session.
Parameters
----------
session_id : str
uuid for the session; this method must be given a session_id
name : str
the .ipynb notebook name that started the session
path : str
the path to the named notebook
kernel_id : str
a uuid for the kernel associated with this session
Returns
-------
model : dict
a dictionary of the session model
"""
self.cursor.execute("INSERT INTO session VALUES (?,?,?,?)",
(session_id, name, path, kernel_id)
)
return self.get_session(session_id=session_id)
def get_session(self, **kwargs):
"""Returns the model for a particular session.
Takes a keyword argument and searches for the value in the session
database, then returns the rest of the session's info.
Parameters
----------
**kwargs : keyword argument
must be given one of the keywords and values from the session database
(i.e. session_id, name, path, kernel_id)
Returns
-------
model : dict
returns a dictionary that includes all the information from the
session described by the kwarg.
"""
if not kwargs:
raise TypeError("must specify a column to query")
conditions = []
for column in kwargs.keys():
if column not in self._columns:
raise TypeError("No such column: %r", column)
conditions.append("%s=?" % column)
query = "SELECT * FROM session WHERE %s" % (' AND '.join(conditions))
self.cursor.execute(query, list(kwargs.values()))
model = self.cursor.fetchone()
if model is None:
q = []
for key, value in kwargs.items():
q.append("%s=%r" % (key, value))
raise web.HTTPError(404, u'Session not found: %s' % (', '.join(q)))
return model
def update_session(self, session_id, **kwargs):
"""Updates the values in the session database.
Changes the values of the session with the given session_id
with the values from the keyword arguments.
Parameters
----------
session_id : str
a uuid that identifies a session in the sqlite3 database
**kwargs : str
the key must correspond to a column title in session database,
and the value replaces the current value in the session
with session_id.
"""
self.get_session(session_id=session_id)
if not kwargs:
# no changes
return
sets = []
for column in kwargs.keys():
if column not in self._columns:
raise TypeError("No such column: %r" % column)
sets.append("%s=?" % column)
query = "UPDATE session SET %s WHERE session_id=?" % (', '.join(sets))
self.cursor.execute(query, list(kwargs.values()) + [session_id])
@staticmethod
def row_factory(cursor, row):
"""Takes sqlite database session row and turns it into a dictionary"""
row = sqlite3.Row(cursor, row)
model = {
'id': row['session_id'],
'notebook': {
'name': row['name'],
'path': row['path']
},
'kernel': {
'id': row['kernel_id'],
}
}
return model
def list_sessions(self):
"""Returns a list of dictionaries containing all the information from
the session database"""
c = self.cursor.execute("SELECT * FROM session")
return list(c.fetchall())
def delete_session(self, session_id):
"""Deletes the row in the session database with given session_id"""
# Check that session exists before deleting
self.get_session(session_id=session_id)
self.cursor.execute("DELETE FROM session WHERE session_id=?", (session_id,))
|