A multi-user server for Jupyter notebooks that provides authentication, spawning, and proxying for multiple users simultaneously
—
JupyterHub uses SQLAlchemy ORM for database operations, providing persistent storage for users, servers, groups, roles, services, and authentication state. The database models support the full JupyterHub feature set including role-based access control, server sharing, and OAuth integration.
Database models for managing users and their associated data.
class User(Base):
"""
Database model for JupyterHub users.
Represents a user account with servers, groups, roles, and tokens.
"""
# Primary attributes
id: int # Primary key
name: str # Username (unique)
admin: bool # Whether user is an admin
created: datetime # When user was created
last_activity: datetime # Last recorded activity
# Authentication state
cookie_id: str # Cookie identifier
state: Dict[str, Any] # Arbitrary state dictionary
encrypted_auth_state: bytes # Encrypted authentication state
# Relationships
servers: List[Server] # User's servers
api_tokens: List[APIToken] # User's API tokens
oauth_tokens: List[APIToken] # OAuth tokens
groups: List[Group] # Groups user belongs to
roles: List[Role] # Roles assigned to user
def new_api_token(self, token=None, roles=None, scopes=None, note=None, expires_in=None):
"""
Create a new API token for this user.
Args:
token: Token string (generated if None)
roles: List of role names
scopes: List of scopes
note: Description of token
expires_in: Expiration time in seconds
Returns:
APIToken object
"""
def get_server(self, name=''):
"""
Get a named server for this user.
Args:
name: Server name (empty string for default server)
Returns:
Server object or None
"""
@property
def escaped_name(self) -> str:
"""URL-escaped username for safe use in URLs"""
@property
def auth_state(self) -> Dict[str, Any]:
"""Decrypted authentication state"""
class Server(Base):
"""
Database model for user notebook servers.
Represents a single-user notebook server instance.
"""
# Primary attributes
id: int # Primary key
name: str # Server name (empty for default)
url: str # Server URL
bind_url: str # Internal bind URL
last_activity: datetime # Last recorded activity
started: datetime # When server was started
base_url: str # Base URL prefix
# State and configuration
state: Dict[str, Any] # Spawner state
# Relationships
user_id: int # Foreign key to User
user: User # User who owns this server
oauth_tokens: List[APIToken] # OAuth tokens for this server
@property
def ready(self) -> bool:
"""Whether server is ready to accept connections"""
@property
def pending(self) -> bool:
"""Whether server is starting up"""
def stop(self):
"""Mark server as stopped"""Models for organizing users and managing permissions.
class Group(Base):
"""
Database model for user groups.
Groups provide a way to organize users and assign permissions.
"""
# Primary attributes
id: int # Primary key
name: str # Group name (unique)
description: str # Group description
# Relationships
users: List[User] # Users in this group
roles: List[Role] # Roles assigned to this group
@property
def member_names(self) -> List[str]:
"""List of usernames in this group"""
class Role(Base):
"""
Database model for roles in the RBAC system.
Roles define collections of scopes that can be assigned to users,
groups, services, and tokens.
"""
# Primary attributes
id: int # Primary key
name: str # Role name (unique)
description: str # Role description
# Permissions
scopes: List[str] # List of scope strings
# Relationships
users: List[User] # Users with this role
groups: List[Group] # Groups with this role
services: List[Service] # Services with this role
tokens: List[APIToken] # Tokens with this role
@classmethod
def find(cls, db, name):
"""
Find role by name.
Args:
db: Database session
name: Role name
Returns:
Role object or None
"""Models for external services and authentication tokens.
class Service(Base):
"""
Database model for JupyterHub services.
Services are external applications that integrate with JupyterHub.
"""
# Primary attributes
id: int # Primary key
name: str # Service name (unique)
admin: bool # Whether service has admin privileges
url: str # Service URL
prefix: str # URL prefix for routing
pid: int # Process ID (for managed services)
# Relationships
api_tokens: List[APIToken] # Service API tokens
oauth_tokens: List[APIToken] # OAuth tokens
roles: List[Role] # Roles assigned to service
oauth_client: OAuthClient # OAuth client info
class APIToken(Base):
"""
Database model for API authentication tokens.
Tokens provide API access for users, services, and OAuth clients.
"""
# Primary attributes
id: int # Primary key
hashed: str # Hashed token value
prefix: str # Token prefix (for identification)
created: datetime # When token was created
last_activity: datetime # Last token usage
expires_at: datetime # Token expiration (optional)
note: str # Description/note about token
# Relationships
user_id: int # Owner user (optional)
user: User # User who owns token
service_id: int # Owner service (optional)
service: Service # Service that owns token
oauth_client_id: str # OAuth client (optional)
oauth_client: OAuthClient # OAuth client
roles: List[Role] # Roles assigned to token
@property
def scopes(self) -> List[str]:
"""All scopes granted to this token"""
def match(self, token):
"""
Check if provided token matches this record.
Args:
token: Token string to check
Returns:
True if token matches
"""Models supporting OAuth 2.0 integration for external applications.
class OAuthClient(Base):
"""
Database model for OAuth client applications.
OAuth clients can obtain tokens to access JupyterHub APIs.
"""
# Primary attributes
id: str # Client ID (primary key)
identifier: str # Client identifier
secret: str # Client secret (hashed)
description: str # Client description
redirect_uri: str # OAuth redirect URI
allowed_scopes: List[str] # Scopes client can request
# Relationships
tokens: List[APIToken] # Tokens issued to this client
codes: List[OAuthCode] # Authorization codes
def check_secret(self, secret):
"""
Verify client secret.
Args:
secret: Secret to verify
Returns:
True if secret matches
"""
class OAuthCode(Base):
"""
Database model for OAuth authorization codes.
Temporary codes used in OAuth authorization flow.
"""
# Primary attributes
id: int # Primary key
client_id: str # OAuth client ID
code: str # Authorization code
expires_at: datetime # Code expiration
redirect_uri: str # Redirect URI
session_id: str # Session identifier
# Relationships
client: OAuthClient # OAuth client
user_id: int # User who authorized
user: User # Authorizing userModels supporting the server sharing feature.
class Share(Base):
"""
Database model for server shares.
Represents a shared server that can be accessed by multiple users.
"""
# Primary attributes
id: int # Primary key
user_id: int # Owner user ID
server_name: str # Shared server name
# Relationships
user: User # User who owns the shared server
codes: List[ShareCode] # Access codes for this share
@property
def server(self) -> Server:
"""The shared server object"""
class ShareCode(Base):
"""
Database model for server share access codes.
Codes that allow access to shared servers.
"""
# Primary attributes
id: int # Primary key
code: str # Access code
created: datetime # When code was created
last_used: datetime # Last time code was used
accept_count: int # Number of times code was used
expires_at: datetime # Code expiration (optional)
# Relationships
share_id: int # Foreign key to Share
share: Share # The share this code belongs tofrom jupyterhub.orm import User, Server, Group, Role
from sqlalchemy.orm import sessionmaker
# Database session example
Session = sessionmaker()
db = Session()
# Create a new user
user = User(name='alice', admin=False)
db.add(user)
db.commit()
# Find user by name
user = db.query(User).filter(User.name == 'alice').first()
# Create API token for user
token = user.new_api_token(
note='API access token',
scopes=['read:users', 'servers']
)
db.commit()
# Query user's servers
servers = db.query(Server).filter(Server.user == user).all()# Create group
group = Group(name='students', description='Student users')
db.add(group)
# Add users to group
alice = db.query(User).filter(User.name == 'alice').first()
bob = db.query(User).filter(User.name == 'bob').first()
group.users.extend([alice, bob])
# Create role and assign to group
role = Role(
name='student-role',
description='Basic student permissions',
scopes=['self', 'servers']
)
group.roles.append(role)
db.commit()# Register external service
service = Service(
name='announcement-service',
admin=False,
url='http://localhost:8001',
prefix='/announcements'
)
db.add(service)
# Create API token for service
token = service.new_api_token(
scopes=['read:users', 'read:servers'],
note='Service API access'
)
db.commit()# Register OAuth client
client = OAuthClient(
id='my-app',
identifier='my-app-client',
description='My Application',
redirect_uri='http://localhost:3000/callback',
allowed_scopes=['read:users', 'read:servers']
)
client.secret = 'hashed-secret-value'
db.add(client)
db.commit()# jupyterhub_config.py
# SQLite (default)
c.JupyterHub.db_url = 'sqlite:///jupyterhub.sqlite'
# PostgreSQL
c.JupyterHub.db_url = 'postgresql://user:password@localhost/jupyterhub'
# MySQL
c.JupyterHub.db_url = 'mysql+pymysql://user:password@localhost/jupyterhub'# Database upgrade command
# jupyterhub upgrade-db
# Programmatic upgrade
from jupyterhub.dbutil import upgrade_if_needed
upgrade_if_needed(db_url, log=app.log)# Common query patterns
from jupyterhub.orm import User, Server, Group
# Active users with servers
active_users = db.query(User).join(Server).filter(
Server.last_activity > cutoff_date
).all()
# Admin users
admins = db.query(User).filter(User.admin == True).all()
# Users in specific group
group_users = db.query(User).join(User.groups).filter(
Group.name == 'students'
).all()
# Expired tokens
expired = db.query(APIToken).filter(
APIToken.expires_at < datetime.utcnow()
).all()Install with Tessl CLI
npx tessl i tessl/pypi-jupyterhub