Authentication with Pocketbase

[thickmn]

Here's a take on auth with Pocketbase and a newer async SDK here.

"""
- Download the pocketbase .exe and place in the root directory https://pocketbase.io/docs/ 

- pip install pocketbase-async

- Run the following command in the terminal to start the pocketbase server:
        pocketbase serve --http=127.0.0.1:8123
        
- visit http://127.0.0.1:8123/_/ and register an admin account

- your 'users' collection is created by default
"""

from typing import Optional

from fastapi import Request
from fastapi.responses import RedirectResponse
from starlette.middleware.base import BaseHTTPMiddleware

from nicegui import Client, app, ui
from pocketbase import PocketBase

POCKETBASE_COLLECTION_NAME = 'users'

pb_client = PocketBase("http://localhost:8123")

unrestricted_page_routes = {'/login', '/register'}


class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if not pb_client._inners.auth._token:
            if request.url.path in Client.page_routes.values() and request.url.path not in unrestricted_page_routes:
                app.storage.user['referrer_path'] = request.url.path
                return RedirectResponse('/login')
        return await call_next(request)
    

app.add_middleware(AuthMiddleware)

@ui.page('/')
def main_page() -> None:
    with ui.column().classes('absolute-center items-center'):
        ui.label(f"Hello {app.storage.user.get('username')}").classes('text-2xl')
        ui.button(on_click=logout, icon='logout').props('outline round')
        

@ui.page('/login')
async def login() -> Optional[RedirectResponse]:
    async def try_login() -> None:
        try:
            await pb_client.collection(POCKETBASE_COLLECTION_NAME).auth.with_password(username.value, password.value)
            app.storage.user.update({'username': username.value, 'authenticated': True})
            ui.navigate.to(app.storage.user.get('referrer_path', '/')) 
        except Exception:
            ui.notify('Wrong username or password', color='negative')

    if pb_client._inners.auth._token:
        return RedirectResponse('/')

    with ui.card().classes('absolute-center'):
        username = ui.input('Username').on('keydown.enter', lambda: try_login())
        password = ui.input('Password', password=True, password_toggle_button=True).on('keydown.enter', lambda: try_login())
        ui.button('Log in', on_click=lambda: try_login())
        ui.link('Register', '/register')
    return None


@ui.page('/register')
async def register() -> Optional[RedirectResponse]:
    async def try_register() -> None:
        try:
            await pb_client.collection(POCKETBASE_COLLECTION_NAME).create(
                {
                    "username": username.value,
                    "email": email.value,
                    "password": password.value,
                    "passwordConfirm": confirm_password.value,
                    "verified": False,
                }
            )
            ui.notify('Registration successful. Please log in.', color='positive')
            ui.navigate.to('/login')
        except Exception as e:
            ui.notify(str(e), color='negative')

    if pb_client._inners.auth._token:
        return RedirectResponse('/')

    with ui.card().classes('absolute-center'):
        username = ui.input('Username').on('keydown.enter', lambda: try_register())
        email = ui.input('Email').on('keydown.enter', lambda: try_register())
        password = ui.input('Password', password=True, password_toggle_button=True).on('keydown.enter', lambda: try_register())
        confirm_password = ui.input('Confirm Password', password=True, password_toggle_button=True).on('keydown.enter', lambda: try_register())
        ui.button('Register', on_click=lambda: try_register())
    return None


async def logout() -> None:
    pb_client._inners.auth.clean()
    return RedirectResponse('/login')

ui.run(storage_secret="TEST")

[rodja]

Very cool. Thanks for sharing. Why do you need to access _inners.auth._token? Is there not a public API to do things like that?

[thickmn]

I’m not sure why I stuck with that, you can likely just get a collection object or .auth object and access user fields.

[thickmn]

Very cool. Thanks for sharing. Why do you need to access _inners.auth._token? Is there not a public API to do things like that?

Finally started implementing this authentication flow and found that the response object has a token after authenticating

@ui.page('/login')
async def login() -> Optional[RedirectResponse]:
    async def try_login() -> None:
        try:
            auth_response = await pb_client.collection(POCKETBASE_USER_COLLECTION).auth.with_password(username.value, password.value)    
            app.storage.user['token'] = auth_response['token'] 
            ui.navigate.to(app.storage.user.get('referrer_path', '/'))  # go back to where the user wanted to go
        except Exception:
            ui.notify('Wrong username or password', color='negative')

    if app.storage.user.get('token'):
        return RedirectResponse('/')

    with ui.card().classes('absolute-center'):
        username = ui.input('Username').on('keydown.enter', lambda: try_login())
        password = ui.input('Password', password=True, password_toggle_button=True).on('keydown.enter', lambda: try_login())
        ui.button('Log in', on_click=lambda: try_login())
        ui.link('Register', '/register')
    return None