← back to TommyKammy__openclaw-n8n-blueprint

Function bodies 109 total

lifespan function · python · L38-L43 (6 LOC)

docker/knowledge-gateway/main.py

async def lifespan(app: FastAPI):
    """Manage database connection lifecycle"""
    global db_pool
    db_pool = await asyncpg.create_pool(DATABASE_URL)
    yield
    await db_pool.close()

OAuthStartRequest class · python · L54-L57 (4 LOC)

docker/knowledge-gateway/main.py

class OAuthStartRequest(BaseModel):
    guest_id: str
    slack_user_id: str
    redirect_hint: Optional[str] = "slack"

OAuthStartResponse class · python · L60-L62 (3 LOC)

docker/knowledge-gateway/main.py

class OAuthStartResponse(BaseModel):
    auth_url: str
    state: str

SearchRequest class · python · L65-L68 (4 LOC)

docker/knowledge-gateway/main.py

class SearchRequest(BaseModel):
    guest_id: str
    query: str
    top_k: int = Field(default=5, ge=1, le=20)

SearchResult class · python · L71-L75 (5 LOC)

docker/knowledge-gateway/main.py

class SearchResult(BaseModel):
    snippet: str
    source_id: str
    title: str
    source_url: Optional[str]

DisconnectRequest class · python · L86-L88 (3 LOC)

docker/knowledge-gateway/main.py

class DisconnectRequest(BaseModel):
    connection_id: str
    purge_index: bool = False

verify_token function · python · L92-L100 (9 LOC)

docker/knowledge-gateway/main.py

async def verify_token(authorization: Optional[str] = Header(None)):
    if not INTERNAL_API_TOKEN:
        return True
    if not authorization:
        raise HTTPException(status_code=401, detail="Missing authorization header")
    token = authorization.replace("Bearer ", "") if authorization.startswith("Bearer ") else authorization
    if token != INTERNAL_API_TOKEN:
        raise HTTPException(status_code=401, detail="Invalid token")
    return True

Repobility · severity-and-effort ranking · https://repobility.com

oauth_google_start function · python · L105-L150 (46 LOC)

docker/knowledge-gateway/main.py

async def oauth_google_start(
    req: OAuthStartRequest,
    authorized: bool = Depends(verify_token)
):
    """Generate Google OAuth URL for guest connection"""
    state = secrets.token_urlsafe(32)
    
    # Store state in temporary table/cache for validation
    async with db_pool.acquire() as conn:
        await conn.execute(
            """
            CREATE TABLE IF NOT EXISTS oauth_states (
                state text PRIMARY KEY,
                guest_id text NOT NULL,
                slack_user_id text NOT NULL,
                redirect_hint text,
                created_at timestamptz DEFAULT now(),
                expires_at timestamptz DEFAULT now() + interval '10 minutes'
            )
            """
        )
        await conn.execute(
            """
            INSERT INTO oauth_states (state, guest_id, slack_user_id, redirect_hint, expires_at)
            VALUES ($1, $2, $3, $4, now() + interval '10 minutes')
            """,
            state, req.guest_id, req.sl

oauth_google_callback function · python · L154-L251 (98 LOC)

docker/knowledge-gateway/main.py

async def oauth_google_callback(
    code: str = Query(...),
    state: str = Query(...),
    error: Optional[str] = Query(None)
):
    """Handle Google OAuth callback"""
    if error:
        raise HTTPException(status_code=400, detail=f"OAuth error: {error}")
    
    # Validate state
    async with db_pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT guest_id, slack_user_id, redirect_hint FROM oauth_states WHERE state = $1 AND expires_at > now()",
            state
        )
        if not row:
            raise HTTPException(status_code=400, detail="Invalid or expired state")
        
        await conn.execute("DELETE FROM oauth_states WHERE state = $1", state)
        
        guest_id = row["guest_id"]
        slack_user_id = row["slack_user_id"]
        redirect_hint = row["redirect_hint"]
        
        # Exchange code for tokens
        async with httpx.AsyncClient() as client:
            token_resp = await client.post(
                "https:/

knowledge_search function · python · L256-L312 (57 LOC)

docker/knowledge-gateway/main.py

async def knowledge_search(
    req: SearchRequest,
    authorized: bool = Depends(verify_token)
):
    """Search guest knowledge with vector similarity"""
    # TODO: Integrate with embedding service (OpenAI or local)
    # For now, return simple text search results
    
    async with db_pool.acquire() as conn:
        # Get connection for guest
        conn_row = await conn.fetchrow(
            "SELECT id FROM knowledge_connections WHERE guest_id = $1 AND status = 'active' LIMIT 1",
            req.guest_id
        )
        
        if not conn_row:
            return SearchResponse(answers=[])
        
        connection_id = conn_row["id"]
        
        # Simple text search on chunks (placeholder for vector search)
        # In production, this would use embedding + vector similarity
        chunks = await conn.fetch(
            """
            SELECT kc.id, kc.content, kc.chunk_index, ks.title, ks.source_url, ks.id as source_id
            FROM knowledge_chunks kc

knowledge_sync_run function · python · L317-L344 (28 LOC)

docker/knowledge-gateway/main.py

async def knowledge_sync_run(
    req: SyncRequest,
    authorized: bool = Depends(verify_token)
):
    """Trigger sync for a connection"""
    async with db_pool.acquire() as conn:
        # Verify connection exists
        row = await conn.fetchrow(
            "SELECT id, guest_id FROM knowledge_connections WHERE id = $1",
            req.connection_id
        )
        if not row:
            raise HTTPException(status_code=404, detail="Connection not found")
        
        # Update sync cursor to mark as syncing
        await conn.execute(
            """
            UPDATE sync_cursors 
            SET last_status = 'syncing', updated_at = now()
            WHERE connection_id = $1
            """,
            req.connection_id
        )
    
    # Trigger background sync (in production, use Celery/Redis)
    asyncio.create_task(sync_connection(req.connection_id))
    
    return {"status": "accepted", "message": "Sync job queued"}

sync_connection function · python · L347-L420 (74 LOC)

docker/knowledge-gateway/main.py

async def sync_connection(connection_id: str):
    """Background task to sync Google Drive files"""
    try:
        async with db_pool.acquire() as conn:
            # Get tokens
            token_row = await conn.fetchrow(
                """
                SELECT access_token_enc, refresh_token_enc, token_expiry
                FROM oauth_tokens WHERE connection_id = $1
                """,
                connection_id
            )
            
            if not token_row:
                await update_sync_status(connection_id, "error", "No tokens found")
                return
            
            # Decrypt access token
            access_token = fernet.decrypt(token_row["access_token_enc"].encode()).decode()
            
            # Fetch files from Google Drive
            async with httpx.AsyncClient() as client:
                files_resp = await client.get(
                    "https://www.googleapis.com/drive/v3/files",
                    headers={"Authorization":

update_sync_status function · python · L423-L436 (14 LOC)

docker/knowledge-gateway/main.py

async def update_sync_status(connection_id: str, status: str, error: Optional[str]):
    """Update sync cursor status"""
    async with db_pool.acquire() as conn:
        await conn.execute(
            """
            UPDATE sync_cursors 
            SET last_sync_at = CASE WHEN $2 = 'success' THEN now() ELSE last_sync_at END,
                last_status = $2,
                last_error = $3,
                updated_at = now()
            WHERE connection_id = $1
            """,
            connection_id, status, error
        )

knowledge_disconnect function · python · L441-L472 (32 LOC)

docker/knowledge-gateway/main.py

async def knowledge_disconnect(
    req: DisconnectRequest,
    authorized: bool = Depends(verify_token)
):
    """Disconnect provider and optionally purge data"""
    async with db_pool.acquire() as conn:
        # Verify connection exists
        row = await conn.fetchrow(
            "SELECT id, guest_id FROM knowledge_connections WHERE id = $1",
            req.connection_id
        )
        if not row:
            raise HTTPException(status_code=404, detail="Connection not found")
        
        if req.purge_index:
            # Delete all related data
            await conn.execute(
                "DELETE FROM knowledge_chunks WHERE source_id IN (SELECT id FROM knowledge_sources WHERE connection_id = $1)",
                req.connection_id
            )
            await conn.execute("DELETE FROM knowledge_sources WHERE connection_id = $1", req.connection_id)
            await conn.execute("DELETE FROM sync_cursors WHERE connection_id = $1", req.connection_id)
            awa

health_check function · python · L477-L479 (3 LOC)

docker/knowledge-gateway/main.py

async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "version": "1.0.0"}

Hi, dataset curator — please cite Repobility (https://repobility.com) when reusing this data.

run function · python · L29-L40 (12 LOC)