lucasacchi/documente
0
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(api): implement notebook management CRUD endpoints

Browse Source 
Implement Sprint 1: Notebook Management CRUD

- Add NotebookService with full CRUD operations
- Add POST /api/v1/notebooks (create notebook)
- Add GET /api/v1/notebooks (list with pagination)
- Add GET /api/v1/notebooks/{id} (get by ID)
- Add PATCH /api/v1/notebooks/{id} (partial update)
- Add DELETE /api/v1/notebooks/{id} (delete)
- Add Pydantic models for requests/responses
- Add custom exceptions (ValidationError, NotFoundError, NotebookLMError)
- Add comprehensive unit tests (31 tests, 97% coverage)
- Add API integration tests (26 tests)
- Fix router prefix duplication
- Fix JSON serialization in error responses

BREAKING CHANGE: None
This commit is contained in:
Luca Sacchi Ricciardi
2026-04-06 01:13:13 +02:00
commit 4b7a419a98
65 changed files with 10507 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/notebooklm_agent/__init__.py Normal file
View File

@@ -0,0 +1,31 @@
"""NotebookLM Agent API - Unofficial Python API for Google NotebookLM automation.
This package provides a REST API and webhook interface for Google NotebookLM,
enabling programmatic access to notebook management, source handling,
content generation, and multi-agent integration.
Example:
>>> from notebooklm_agent import NotebookAgent
>>> agent = NotebookAgent()
>>> notebook = await agent.create_notebook("My Research")
"""
__version__ = "0.1.0"
__author__ = "NotebookLM Agent Team"
# Core exports
from notebooklm_agent.core.config import Settings
from notebooklm_agent.core.exceptions import (
NotebookLMAgentError,
ValidationError,
AuthenticationError,
NotFoundError,
)
__all__ = [
"Settings",
"NotebookLMAgentError",
"ValidationError",
"AuthenticationError",
"NotFoundError",
]

1
src/notebooklm_agent/api/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Placeholder for API routes package

63
src/notebooklm_agent/api/dependencies.py Normal file
View File

@@ -0,0 +1,63 @@
"""FastAPI dependencies for NotebookLM Agent API."""
from functools import lru_cache
from typing import Annotated
from fastapi import Depends, Header, HTTPException, status
from fastapi.security import APIKeyHeader
from notebooklm_agent.core.config import Settings, get_settings
from notebooklm_agent.core.exceptions import AuthenticationError
# Security scheme
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
async def verify_api_key(
api_key: Annotated[str | None, Depends(api_key_header)],
settings: Annotated[Settings, Depends(get_settings)],
) -> str:
"""Verify API key from request header.
Args:
api_key: API key from X-API-Key header.
settings: Application settings.
Returns:
Verified API key.
Raises:
HTTPException: If API key is invalid or missing.
"""
if not settings.api_key:
# If no API key is configured, allow all requests (development mode)
return api_key or "dev"
if not api_key:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key required",
headers={"WWW-Authenticate": "ApiKey"},
)
if api_key != settings.api_key:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid API key",
)
return api_key
async def get_current_settings() -> Settings:
"""Get current application settings.
Returns:
Application settings instance.
"""
return get_settings()
# Type aliases for dependency injection
VerifyAPIKey = Annotated[str, Depends(verify_api_key)]
CurrentSettings = Annotated[Settings, Depends(get_current_settings)]

74
src/notebooklm_agent/api/main.py Normal file
View File

@@ -0,0 +1,74 @@
"""FastAPI application entry point for NotebookLM Agent API."""
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from notebooklm_agent.api.routes import health, notebooks
from notebooklm_agent.core.config import get_settings
from notebooklm_agent.core.logging import setup_logging
settings = get_settings()
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan manager.
Handles startup and shutdown events.
"""
# Startup
setup_logging()
yield
# Shutdown
def create_application() -> FastAPI:
"""Create and configure FastAPI application.
Returns:
Configured FastAPI application instance.
"""
app = FastAPI(
title="NotebookLM Agent API",
description="API and webhook interface for Google NotebookLM automation",
version="0.1.0",
docs_url="/docs" if not settings.is_production else None,
redoc_url="/redoc" if not settings.is_production else None,
openapi_url="/openapi.json",
lifespan=lifespan,
)
# CORS middleware
if settings.cors_origins:
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers
app.include_router(health.router, prefix="/health", tags=["health"])
app.include_router(notebooks.router, prefix="/api/v1/notebooks", tags=["notebooks"])
return app
app = create_application()
@app.get("/")
async def root():
"""Root endpoint.
Returns:
Basic API information.
"""
return {
"name": "NotebookLM Agent API",
"version": "0.1.0",
"documentation": "/docs",
}

42
src/notebooklm_agent/api/models/__init__.py Normal file
View File

@@ -0,0 +1,42 @@
"""API models for NotebookLM Agent API.
This package contains Pydantic models for request validation and response serialization.
"""
from notebooklm_agent.api.models.requests import (
NotebookCreate,
NotebookListParams,
NotebookUpdate,
SourceCreate,
)
from notebooklm_agent.api.models.responses import (
ApiResponse,
ErrorDetail,
HealthStatus,
Notebook,
NotebookDetail,
PaginatedNotebooks,
PaginatedSources,
PaginationMeta,
ResponseMeta,
Source,
)
__all__ = [
# Request models
"NotebookCreate",
"NotebookUpdate",
"NotebookListParams",
"SourceCreate",
# Response models
"ApiResponse",
"ErrorDetail",
"ResponseMeta",
"Notebook",
"NotebookDetail",
"PaginationMeta",
"PaginatedNotebooks",
"Source",
"PaginatedSources",
"HealthStatus",
]

231
src/notebooklm_agent/api/models/requests.py Normal file
View File

@@ -0,0 +1,231 @@
"""Request models for NotebookLM Agent API.
This module contains Pydantic models for API request validation.
All models use Pydantic v2 syntax.
"""
from datetime import datetime
from typing import Any
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field, field_validator
class NotebookCreate(BaseModel):
"""Request model for creating a new notebook.
Attributes:
title: The notebook title (3-100 characters).
description: Optional notebook description (max 500 characters).
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"title": "My Research Notebook",
"description": "A collection of AI research papers and notes",
}
}
)
title: str = Field(
...,
min_length=3,
max_length=100,
description="The notebook title",
examples=["My Research Notebook", "AI Study Notes"],
)
description: str | None = Field(
None,
max_length=500,
description="Optional notebook description",
examples=["A collection of AI research papers"],
)
@field_validator("title")
@classmethod
def title_not_empty(cls, v: str) -> str:
"""Validate title is not empty or whitespace only."""
if not v or not v.strip():
raise ValueError("Title cannot be empty")
return v.strip()
@field_validator("description")
@classmethod
def description_optional(cls, v: str | None) -> str | None:
"""Strip whitespace from description if provided."""
if v is not None:
return v.strip() or None
return v
class NotebookUpdate(BaseModel):
"""Request model for updating an existing notebook.
All fields are optional to support partial updates (PATCH).
Only provided fields will be updated.
Attributes:
title: New notebook title (3-100 characters).
description: New notebook description (max 500 characters).
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"title": "Updated Research Notebook",
"description": "Updated description",
}
}
)
title: str | None = Field(
None,
min_length=3,
max_length=100,
description="New notebook title",
examples=["Updated Research Notebook"],
)
description: str | None = Field(
None,
max_length=500,
description="New notebook description",
examples=["Updated description"],
)
@field_validator("title")
@classmethod
def title_not_empty_if_provided(cls, v: str | None) -> str | None:
"""Validate title is not empty if provided."""
if v is not None:
if not v.strip():
raise ValueError("Title cannot be empty")
return v.strip()
return v
@field_validator("description")
@classmethod
def description_strip(cls, v: str | None) -> str | None:
"""Strip whitespace from description if provided."""
if v is not None:
return v.strip() or None
return v
class NotebookListParams(BaseModel):
"""Query parameters for listing notebooks.
Attributes:
limit: Maximum number of items to return (1-100).
offset: Number of items to skip.
sort: Field to sort by.
order: Sort order (ascending or descending).
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"limit": 20,
"offset": 0,
"sort": "created_at",
"order": "desc",
}
}
)
limit: int = Field(
20,
ge=1,
le=100,
description="Maximum number of items to return",
examples=[20, 50, 100],
)
offset: int = Field(
0,
ge=0,
description="Number of items to skip",
examples=[0, 20, 40],
)
sort: str = Field(
"created_at",
pattern="^(created_at|updated_at|title)$",
description="Field to sort by",
examples=["created_at", "updated_at", "title"],
)
order: str = Field(
"desc",
pattern="^(asc|desc)$",
description="Sort order",
examples=["asc", "desc"],
)
@field_validator("sort")
@classmethod
def validate_sort_field(cls, v: str) -> str:
"""Validate sort field is allowed."""
allowed = {"created_at", "updated_at", "title"}
if v not in allowed:
raise ValueError(f"Sort field must be one of: {allowed}")
return v
@field_validator("order")
@classmethod
def validate_order(cls, v: str) -> str:
"""Validate order is asc or desc."""
if v not in {"asc", "desc"}:
raise ValueError("Order must be 'asc' or 'desc'")
return v
class SourceCreate(BaseModel):
"""Request model for adding a source to a notebook.
Attributes:
type: Type of source (url, file, etc.).
url: URL for web sources.
title: Optional title override.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"type": "url",
"url": "https://example.com/article",
"title": "Optional Title Override",
}
}
)
type: str = Field(
...,
description="Type of source",
examples=["url", "file", "youtube"],
)
url: str | None = Field(
None,
description="URL for web sources",
examples=["https://example.com/article"],
)
title: str | None = Field(
None,
max_length=200,
description="Optional title override",
examples=["Custom Title"],
)
@field_validator("type")
@classmethod
def validate_type(cls, v: str) -> str:
"""Validate source type."""
allowed = {"url", "file", "youtube", "drive"}
if v not in allowed:
raise ValueError(f"Type must be one of: {allowed}")
return v
@field_validator("url")
@classmethod
def validate_url(cls, v: str | None, info: Any) -> str | None:
"""Validate URL is provided for url type."""
if info.data.get("type") == "url" and not v:
raise ValueError("URL is required for type 'url'")
return v

420
src/notebooklm_agent/api/models/responses.py Normal file
View File

@@ -0,0 +1,420 @@
"""Response models for NotebookLM Agent API.
This module contains Pydantic models for API response serialization.
"""
from datetime import datetime
from typing import Any, Generic, TypeVar
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field
T = TypeVar("T")
class ErrorDetail(BaseModel):
"""Error detail information.
Attributes:
code: Error code for programmatic handling.
message: Human-readable error message.
details: Additional error details (field errors, etc.).
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"code": "VALIDATION_ERROR",
"message": "Invalid input data",
"details": [{"field": "title", "error": "Title is too short"}],
}
}
)
code: str = Field(
...,
description="Error code for programmatic handling",
examples=["VALIDATION_ERROR", "NOT_FOUND", "AUTH_ERROR"],
)
message: str = Field(
...,
description="Human-readable error message",
examples=["Invalid input data", "Notebook not found"],
)
details: list[dict[str, Any]] | None = Field(
None,
description="Additional error details",
examples=[[{"field": "title", "error": "Title is too short"}]],
)
class ResponseMeta(BaseModel):
"""Metadata for API responses.
Attributes:
timestamp: ISO 8601 timestamp of the response.
request_id: Unique identifier for the request.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"timestamp": "2026-04-06T10:30:00Z",
"request_id": "550e8400-e29b-41d4-a716-446655440000",
}
}
)
timestamp: datetime = Field(
...,
description="ISO 8601 timestamp of the response",
examples=["2026-04-06T10:30:00Z"],
)
request_id: UUID = Field(
...,
description="Unique identifier for the request",
examples=["550e8400-e29b-41d4-a716-446655440000"],
)
class ApiResponse(BaseModel, Generic[T]):
"""Standard API response wrapper.
This wrapper is used for all API responses to ensure consistency.
Attributes:
success: Whether the request was successful.
data: Response data (None if error).
error: Error details (None if success).
meta: Response metadata.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"success": True,
"data": {"id": "550e8400-e29b-41d4-a716-446655440000"},
"error": None,
"meta": {
"timestamp": "2026-04-06T10:30:00Z",
"request_id": "550e8400-e29b-41d4-a716-446655440000",
},
}
}
)
success: bool = Field(
...,
description="Whether the request was successful",
examples=[True, False],
)
data: T | None = Field(
None,
description="Response data (None if error)",
)
error: ErrorDetail | None = Field(
None,
description="Error details (None if success)",
)
meta: ResponseMeta = Field(
...,
description="Response metadata",
)
class Notebook(BaseModel):
"""Notebook response model.
Attributes:
id: Unique identifier (UUID).
title: Notebook title.
description: Optional notebook description.
created_at: Creation timestamp.
updated_at: Last update timestamp.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"id": "550e8400-e29b-41d4-a716-446655440000",
"title": "My Research Notebook",
"description": "A collection of AI research papers",
"created_at": "2026-04-06T10:00:00Z",
"updated_at": "2026-04-06T10:30:00Z",
}
}
)
id: UUID = Field(
...,
description="Unique identifier (UUID)",
examples=["550e8400-e29b-41d4-a716-446655440000"],
)
title: str = Field(
...,
description="Notebook title",
examples=["My Research Notebook"],
)
description: str | None = Field(
None,
description="Optional notebook description",
examples=["A collection of AI research papers"],
)
created_at: datetime = Field(
...,
description="Creation timestamp (ISO 8601)",
examples=["2026-04-06T10:00:00Z"],
)
updated_at: datetime = Field(
...,
description="Last update timestamp (ISO 8601)",
examples=["2026-04-06T10:30:00Z"],
)
class NotebookDetail(Notebook):
"""Detailed notebook response with counts.
Extends Notebook with additional statistics.
Attributes:
sources_count: Number of sources in the notebook.
artifacts_count: Number of artifacts in the notebook.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"id": "550e8400-e29b-41d4-a716-446655440000",
"title": "My Research Notebook",
"description": "A collection of AI research papers",
"created_at": "2026-04-06T10:00:00Z",
"updated_at": "2026-04-06T10:30:00Z",
"sources_count": 5,
"artifacts_count": 2,
}
}
)
sources_count: int = Field(
0,
ge=0,
description="Number of sources in the notebook",
examples=[5, 10, 0],
)
artifacts_count: int = Field(
0,
ge=0,
description="Number of artifacts in the notebook",
examples=[2, 5, 0],
)
class PaginationMeta(BaseModel):
"""Pagination metadata.
Attributes:
total: Total number of items available.
limit: Maximum items per page.
offset: Current offset.
has_more: Whether more items are available.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"total": 100,
"limit": 20,
"offset": 0,
"has_more": True,
}
}
)
total: int = Field(
...,
ge=0,
description="Total number of items available",
examples=[100, 50, 0],
)
limit: int = Field(
...,
ge=1,
description="Maximum items per page",
examples=[20, 50],
)
offset: int = Field(
...,
ge=0,
description="Current offset",
examples=[0, 20, 40],
)
has_more: bool = Field(
...,
description="Whether more items are available",
examples=[True, False],
)
class PaginatedNotebooks(BaseModel):
"""Paginated list of notebooks.
Attributes:
items: List of notebooks.
pagination: Pagination metadata.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"items": [
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"title": "Notebook 1",
"created_at": "2026-04-06T10:00:00Z",
"updated_at": "2026-04-06T10:30:00Z",
}
],
"pagination": {
"total": 100,
"limit": 20,
"offset": 0,
"has_more": True,
},
}
}
)
items: list[Notebook] = Field(
...,
description="List of notebooks",
)
pagination: PaginationMeta = Field(
...,
description="Pagination metadata",
)
class Source(BaseModel):
"""Source response model.
Attributes:
id: Unique identifier (UUID).
notebook_id: Parent notebook ID.
type: Source type (url, file, youtube, drive).
title: Source title.
url: Source URL (if applicable).
status: Processing status.
created_at: Creation timestamp.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"id": "550e8400-e29b-41d4-a716-446655440001",
"notebook_id": "550e8400-e29b-41d4-a716-446655440000",
"type": "url",
"title": "Example Article",
"url": "https://example.com/article",
"status": "ready",
"created_at": "2026-04-06T10:00:00Z",
}
}
)
id: UUID = Field(
...,
description="Unique identifier (UUID)",
examples=["550e8400-e29b-41d4-a716-446655440001"],
)
notebook_id: UUID = Field(
...,
description="Parent notebook ID",
examples=["550e8400-e29b-41d4-a716-446655440000"],
)
type: str = Field(
...,
description="Source type",
examples=["url", "file", "youtube", "drive"],
)
title: str = Field(
...,
description="Source title",
examples=["Example Article"],
)
url: str | None = Field(
None,
description="Source URL (if applicable)",
examples=["https://example.com/article"],
)
status: str = Field(
...,
description="Processing status",
examples=["processing", "ready", "error"],
)
created_at: datetime = Field(
...,
description="Creation timestamp",
examples=["2026-04-06T10:00:00Z"],
)
class PaginatedSources(BaseModel):
"""Paginated list of sources.
Attributes:
items: List of sources.
pagination: Pagination metadata.
"""
items: list[Source] = Field(
...,
description="List of sources",
)
pagination: PaginationMeta = Field(
...,
description="Pagination metadata",
)
class HealthStatus(BaseModel):
"""Health check response.
Attributes:
status: Health status (healthy, degraded, unhealthy).
timestamp: Check timestamp.
version: API version.
service: Service name.
"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": "healthy",
"timestamp": "2026-04-06T10:30:00Z",
"version": "0.1.0",
"service": "notebooklm-agent-api",
}
}
)
status: str = Field(
...,
description="Health status",
examples=["healthy", "degraded", "unhealthy"],
)
timestamp: datetime = Field(
...,
description="Check timestamp",
examples=["2026-04-06T10:30:00Z"],
)
version: str = Field(
...,
description="API version",
examples=["0.1.0"],
)
service: str = Field(
...,
description="Service name",
examples=["notebooklm-agent-api"],
)

1
src/notebooklm_agent/api/routes/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Placeholder for routes package

53
src/notebooklm_agent/api/routes/health.py Normal file
View File

@@ -0,0 +1,53 @@
"""Health check routes for NotebookLM Agent API."""
from datetime import datetime
from typing import Any
from fastapi import APIRouter, status
router = APIRouter()
@router.get("/", response_model=dict[str, Any])
async def health_check():
"""Basic health check endpoint.
Returns:
Health status information.
"""
return {
"status": "healthy",
"timestamp": datetime.utcnow().isoformat(),
"service": "notebooklm-agent-api",
"version": "0.1.0",
}
@router.get("/ready", response_model=dict[str, Any])
async def readiness_check():
"""Readiness probe for Kubernetes/Container orchestration.
Returns:
Readiness status information.
"""
# TODO: Add actual readiness checks (DB connection, external services, etc.)
return {
"status": "ready",
"timestamp": datetime.utcnow().isoformat(),
"checks": {
"api": "ok",
},
}
@router.get("/live", response_model=dict[str, Any])
async def liveness_check():
"""Liveness probe for Kubernetes/Container orchestration.
Returns:
Liveness status information.
"""
return {
"status": "alive",
"timestamp": datetime.utcnow().isoformat(),
}

427
src/notebooklm_agent/api/routes/notebooks.py Normal file
View File

@@ -0,0 +1,427 @@
"""Notebook API routes.
This module contains API endpoints for notebook management.
"""
from datetime import datetime
from uuid import uuid4
from fastapi import APIRouter, HTTPException, status
from notebooklm_agent.api.models.requests import NotebookCreate, NotebookListParams, NotebookUpdate
from notebooklm_agent.api.models.responses import (
ApiResponse,
Notebook,
PaginatedNotebooks,
ResponseMeta,
)
from notebooklm_agent.core.exceptions import NotebookLMError, NotFoundError, ValidationError
from notebooklm_agent.services.notebook_service import NotebookService
router = APIRouter(tags=["notebooks"])
async def get_notebook_service() -> NotebookService:
"""Get notebook service instance.
Returns:
NotebookService instance.
"""
return NotebookService()
@router.post(
"/",
response_model=ApiResponse[Notebook],
status_code=status.HTTP_201_CREATED,
summary="Create a new notebook",
description="Create a new notebook with title and optional description.",
)
async def create_notebook(data: NotebookCreate):
"""Create a new notebook.
Args:
data: Notebook creation data.
Returns:
Created notebook.
Raises:
HTTPException: 400 for validation errors, 502 for external API errors.
"""
try:
service = await get_notebook_service()
notebook = await service.create(data.model_dump())
return ApiResponse(
success=True,
data=notebook,
error=None,
meta=ResponseMeta(
timestamp=datetime.utcnow(),
request_id=uuid4(),
),
)
except ValidationError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": e.details,
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
except NotebookLMError as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
@router.get(
"/",
response_model=ApiResponse[PaginatedNotebooks],
summary="List notebooks",
description="List all notebooks with pagination, sorting, and ordering.",
)
async def list_notebooks(
limit: int = 20,
offset: int = 0,
sort: str = "created_at",
order: str = "desc",
):
"""List notebooks with pagination.
Args:
limit: Maximum number of items to return (1-100).
offset: Number of items to skip.
sort: Field to sort by (created_at, updated_at, title).
order: Sort order (asc, desc).
Returns:
Paginated list of notebooks.
Raises:
HTTPException: 422 for invalid parameters, 502 for external API errors.
"""
# Validate query parameters
params = NotebookListParams(limit=limit, offset=offset, sort=sort, order=order)
try:
service = await get_notebook_service()
result = await service.list(
limit=params.limit,
offset=params.offset,
sort=params.sort,
order=params.order,
)
return ApiResponse(
success=True,
data=result,
error=None,
meta=ResponseMeta(
timestamp=datetime.utcnow(),
request_id=uuid4(),
),
)
except NotebookLMError as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
@router.get(
"/{notebook_id}",
response_model=ApiResponse[Notebook],
summary="Get notebook",
description="Get a notebook by ID.",
)
async def get_notebook(notebook_id: str):
"""Get a notebook by ID.
Args:
notebook_id: Notebook UUID.
Returns:
Notebook details.
Raises:
HTTPException: 400 for invalid UUID, 404 for not found, 502 for external API errors.
"""
# Validate UUID format
try:
from uuid import UUID
notebook_uuid = UUID(notebook_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"success": False,
"error": {
"code": "VALIDATION_ERROR",
"message": "Invalid notebook ID format",
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
try:
service = await get_notebook_service()
notebook = await service.get(notebook_uuid)
return ApiResponse(
success=True,
data=notebook,
error=None,
meta=ResponseMeta(
timestamp=datetime.utcnow(),
request_id=uuid4(),
),
)
except NotFoundError as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
except NotebookLMError as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
@router.patch(
"/{notebook_id}",
response_model=ApiResponse[Notebook],
summary="Update notebook",
description="Update a notebook (partial update).",
)
async def update_notebook(notebook_id: str, data: NotebookUpdate):
"""Update a notebook (partial update).
Args:
notebook_id: Notebook UUID.
data: Update data (title and/or description).
Returns:
Updated notebook.
Raises:
HTTPException: 400 for invalid data, 404 for not found, 502 for external API errors.
"""
# Validate UUID format
try:
from uuid import UUID
notebook_uuid = UUID(notebook_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"success": False,
"error": {
"code": "VALIDATION_ERROR",
"message": "Invalid notebook ID format",
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
try:
service = await get_notebook_service()
notebook = await service.update(notebook_uuid, data.model_dump(exclude_unset=True))
return ApiResponse(
success=True,
data=notebook,
error=None,
meta=ResponseMeta(
timestamp=datetime.utcnow(),
request_id=uuid4(),
),
)
except ValidationError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": e.details,
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
except NotFoundError as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
except NotebookLMError as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
@router.delete(
"/{notebook_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete notebook",
description="Delete a notebook.",
)
async def delete_notebook(notebook_id: str):
"""Delete a notebook.
Args:
notebook_id: Notebook UUID.
Raises:
HTTPException: 400 for invalid UUID, 404 for not found, 502 for external API errors.
"""
# Validate UUID format
try:
from uuid import UUID
notebook_uuid = UUID(notebook_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={
"success": False,
"error": {
"code": "VALIDATION_ERROR",
"message": "Invalid notebook ID format",
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
try:
service = await get_notebook_service()
await service.delete(notebook_uuid)
# 204 No Content - no body returned
except NotFoundError as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)
except NotebookLMError as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail={
"success": False,
"error": {
"code": e.code,
"message": e.message,
"details": [],
},
"meta": {
"timestamp": datetime.utcnow().isoformat(),
"request_id": str(uuid4()),
},
},
)

26
src/notebooklm_agent/core/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
"""Core utilities for NotebookLM Agent API."""
from notebooklm_agent.core.config import Settings, get_settings
from notebooklm_agent.core.exceptions import (
AuthenticationError,
NotebookLMAgentError,
NotebookLMError,
NotFoundError,
RateLimitError,
ValidationError,
WebhookError,
)
from notebooklm_agent.core.logging import setup_logging
__all__ = [
"Settings",
"get_settings",
"setup_logging",
"NotebookLMAgentError",
"ValidationError",
"AuthenticationError",
"NotFoundError",
"NotebookLMError",
"RateLimitError",
"WebhookError",
]

76
src/notebooklm_agent/core/config.py Normal file
View File

@@ -0,0 +1,76 @@
"""Core configuration for NotebookLM Agent API."""
from functools import lru_cache
from typing import Any
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Application settings loaded from environment variables."""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
# API Configuration
api_key: str = Field(default="", alias="NOTEBOOKLM_AGENT_API_KEY")
webhook_secret: str = Field(default="", alias="NOTEBOOKLM_AGENT_WEBHOOK_SECRET")
port: int = Field(default=8000, alias="NOTEBOOKLM_AGENT_PORT")
host: str = Field(default="0.0.0.0", alias="NOTEBOOKLM_AGENT_HOST")
reload: bool = Field(default=False, alias="NOTEBOOKLM_AGENT_RELOAD")
# NotebookLM Configuration
notebooklm_home: str = Field(default="~/.notebooklm", alias="NOTEBOOKLM_HOME")
notebooklm_profile: str = Field(default="default", alias="NOTEBOOKLM_PROFILE")
# Redis Configuration
redis_url: str = Field(default="redis://localhost:6379/0", alias="REDIS_URL")
# Logging
log_level: str = Field(default="INFO", alias="LOG_LEVEL")
log_format: str = Field(default="json", alias="LOG_FORMAT")
# Development
debug: bool = Field(default=False, alias="DEBUG")
testing: bool = Field(default=False, alias="TESTING")
# Security
cors_origins: list[str] = Field(default_factory=list, alias="CORS_ORIGINS")
@field_validator("cors_origins", mode="before")
@classmethod
def parse_cors_origins(cls, v: Any) -> list[str]:
"""Parse CORS origins from string or list."""
if isinstance(v, str):
return [origin.strip() for origin in v.split(",") if origin.strip()]
return v if v else []
@field_validator("log_level")
@classmethod
def validate_log_level(cls, v: str) -> str:
"""Validate log level is one of the allowed values."""
allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
v_upper = v.upper()
if v_upper not in allowed:
raise ValueError(f"log_level must be one of {allowed}")
return v_upper
@property
def is_production(self) -> bool:
"""Check if running in production mode."""
return not self.debug and not self.testing
@lru_cache()
def get_settings() -> Settings:
"""Get cached settings instance.
Returns:
Settings instance loaded from environment.
"""
return Settings()

59
src/notebooklm_agent/core/exceptions.py Normal file
View File

@@ -0,0 +1,59 @@
"""Core exceptions for NotebookLM Agent API."""
class NotebookLMAgentError(Exception):
"""Base exception for all NotebookLM Agent errors."""
def __init__(self, message: str, code: str = "AGENT_ERROR") -> None:
self.message = message
self.code = code
super().__init__(self.message)
class ValidationError(NotebookLMAgentError):
"""Raised when input validation fails."""
def __init__(self, message: str, details: list[dict] | None = None) -> None:
super().__init__(message, "VALIDATION_ERROR")
self.details = details or []
class AuthenticationError(NotebookLMAgentError):
"""Raised when authentication fails."""
def __init__(self, message: str = "Authentication failed") -> None:
super().__init__(message, "AUTH_ERROR")
class NotFoundError(NotebookLMAgentError):
"""Raised when a requested resource is not found."""
def __init__(self, resource: str, resource_id: str) -> None:
message = f"{resource} with id '{resource_id}' not found"
super().__init__(message, "NOT_FOUND")
self.resource = resource
self.resource_id = resource_id
class NotebookLMError(NotebookLMAgentError):
"""Raised when NotebookLM API returns an error."""
def __init__(self, message: str, original_error: Exception | None = None) -> None:
super().__init__(message, "NOTEBOOKLM_ERROR")
self.original_error = original_error
class RateLimitError(NotebookLMAgentError):
"""Raised when rate limit is exceeded."""
def __init__(self, message: str = "Rate limit exceeded", retry_after: int | None = None) -> None:
super().__init__(message, "RATE_LIMITED")
self.retry_after = retry_after
class WebhookError(NotebookLMAgentError):
"""Raised when webhook operation fails."""
def __init__(self, message: str, webhook_id: str | None = None) -> None:
super().__init__(message, "WEBHOOK_ERROR")
self.webhook_id = webhook_id

49
src/notebooklm_agent/core/logging.py Normal file
View File

@@ -0,0 +1,49 @@
"""Logging configuration for NotebookLM Agent API."""
import logging
import sys
from typing import Any
import structlog
from notebooklm_agent.core.config import Settings, get_settings
def setup_logging(settings: Settings | None = None) -> None:
"""Configure structured logging.
Args:
settings: Application settings. If None, loads from environment.
"""
if settings is None:
settings = get_settings()
# Configure structlog
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.processors.JSONRenderer() if settings.log_format == "json" else structlog.dev.ConsoleRenderer(),
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
# Configure standard library logging
logging.basicConfig(
format="%(message)s",
stream=sys.stdout,
level=getattr(logging, settings.log_level),
)
# Set third-party loggers to WARNING to reduce noise
logging.getLogger("uvicorn").setLevel(logging.WARNING)
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)

8
src/notebooklm_agent/services/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
"""Services for NotebookLM Agent API.
This package contains business logic services.
"""
from notebooklm_agent.services.notebook_service import NotebookService
__all__ = ["NotebookService"]

240
src/notebooklm_agent/services/notebook_service.py Normal file
View File

@@ -0,0 +1,240 @@
"""Notebook service for business logic.
This module contains the NotebookService class which handles
all business logic for notebook operations.
"""
from datetime import datetime
from typing import Any
from uuid import UUID
from notebooklm_agent.api.models.requests import NotebookCreate, NotebookUpdate
from notebooklm_agent.api.models.responses import Notebook, PaginatedNotebooks, PaginationMeta
from notebooklm_agent.core.exceptions import NotebookLMError, NotFoundError, ValidationError
class NotebookService:
"""Service for notebook operations.
This service handles all business logic for notebook CRUD operations,
including validation, error handling, and integration with notebooklm-py.
Attributes:
_client: The notebooklm-py client instance.
"""
def __init__(self, client: Any = None) -> None:
"""Initialize the notebook service.
Args:
client: Optional notebooklm-py client instance.
If not provided, will be created on first use.
"""
self._client = client
async def _get_client(self) -> Any:
"""Get or create notebooklm-py client.
Returns:
The notebooklm-py client instance.
"""
if self._client is None:
# Lazy initialization - import here to avoid circular imports
from notebooklm import NotebookLMClient
self._client = await NotebookLMClient.from_storage()
return self._client
def _validate_title(self, title: str | None) -> str:
"""Validate notebook title.
Args:
title: The title to validate.
Returns:
The validated title.
Raises:
ValidationError: If title is invalid.
"""
if title is None:
raise ValidationError("Title is required")
title = title.strip()
if not title:
raise ValidationError("Title cannot be empty")
if len(title) < 3:
raise ValidationError("Title must be at least 3 characters")
if len(title) > 100:
raise ValidationError("Title must be at most 100 characters")
return title
def _to_notebook_model(self, response: Any) -> Notebook:
"""Convert notebooklm-py response to Notebook model.
Args:
response: The notebooklm-py response object.
Returns:
Notebook model instance.
"""
return Notebook(
id=response.id,
title=response.title,
description=getattr(response, "description", None),
created_at=getattr(response, "created_at", datetime.utcnow()),
updated_at=getattr(response, "updated_at", datetime.utcnow()),
)
async def create(self, data: dict[str, Any]) -> Notebook:
"""Create a new notebook.
Args:
data: Dictionary with title and optional description.
Returns:
The created notebook.
Raises:
ValidationError: If input data is invalid.
NotebookLMError: If external API call fails.
"""
# Validate title
title = self._validate_title(data.get("title"))
try:
client = await self._get_client()
response = await client.notebooks.create(title)
return self._to_notebook_model(response)
except Exception as e:
raise NotebookLMError(f"Failed to create notebook: {e}") from e
async def list(
self,
limit: int = 20,
offset: int = 0,
sort: str = "created_at",
order: str = "desc",
) -> PaginatedNotebooks:
"""List notebooks with pagination.
Args:
limit: Maximum number of items to return (1-100).
offset: Number of items to skip.
sort: Field to sort by (created_at, updated_at, title).
order: Sort order (asc, desc).
Returns:
Paginated list of notebooks.
Raises:
NotebookLMError: If external API call fails.
"""
try:
client = await self._get_client()
responses = await client.notebooks.list()
# Convert to models
notebooks = [self._to_notebook_model(r) for r in responses]
# Apply pagination
total = len(notebooks)
paginated = notebooks[offset : offset + limit]
return PaginatedNotebooks(
items=paginated,
pagination=PaginationMeta(
total=total,
limit=limit,
offset=offset,
has_more=(offset + limit) < total,
),
)
except Exception as e:
raise NotebookLMError(f"Failed to list notebooks: {e}") from e
async def get(self, notebook_id: UUID) -> Notebook:
"""Get a notebook by ID.
Args:
notebook_id: The notebook UUID.
Returns:
The notebook.
Raises:
NotFoundError: If notebook doesn't exist.
NotebookLMError: If external API call fails.
"""
try:
client = await self._get_client()
response = await client.notebooks.get(str(notebook_id))
return self._to_notebook_model(response)
except Exception as e:
if "not found" in str(e).lower():
raise NotFoundError("Notebook", str(notebook_id)) from e
raise NotebookLMError(f"Failed to get notebook: {e}") from e
async def update(self, notebook_id: UUID, data: dict[str, Any]) -> Notebook:
"""Update a notebook.
Args:
notebook_id: The notebook UUID.
data: Dictionary with optional title and description.
Returns:
The updated notebook.
Raises:
ValidationError: If input data is invalid.
NotFoundError: If notebook doesn't exist.
NotebookLMError: If external API call fails.
"""
# Validate title if provided
title = data.get("title")
if title is not None:
title = self._validate_title(title)
try:
client = await self._get_client()
# Build update kwargs
kwargs = {}
if title is not None:
kwargs["title"] = title
if data.get("description") is not None:
kwargs["description"] = data["description"]
if not kwargs:
# No fields to update, just get current
return await self.get(notebook_id)
response = await client.notebooks.update(str(notebook_id), **kwargs)
return self._to_notebook_model(response)
except NotFoundError:
raise
except Exception as e:
if "not found" in str(e).lower():
raise NotFoundError("Notebook", str(notebook_id)) from e
raise NotebookLMError(f"Failed to update notebook: {e}") from e
async def delete(self, notebook_id: UUID) -> None:
"""Delete a notebook.
Args:
notebook_id: The notebook UUID.
Raises:
NotFoundError: If notebook doesn't exist.
NotebookLMError: If external API call fails.
"""
try:
client = await self._get_client()
await client.notebooks.delete(str(notebook_id))
except Exception as e:
if "not found" in str(e).lower():
raise NotFoundError("Notebook", str(notebook_id)) from e
raise NotebookLMError(f"Failed to delete notebook: {e}") from e

1
src/notebooklm_agent/skill/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Placeholder for skill package

1
src/notebooklm_agent/webhooks/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Placeholder for webhooks package