Writing an app

uvx --from druks druks create app night_watch
cd druks-night_watch
uv sync
uv run pytest

[project.entry-points."druks.apps"]
night_watch = "druks_night_watch.app:NightWatch"

uv pip install -e /path/to/druks-night_watch

[tool.uv.sources]
druks = { path = "../druks", editable = true }

from druks.apps import App


class NightWatch(App):
    name = "night_watch"
    icon = "telescope"
    description = "Checks repositories after hours."

from druks.workflows import Workflow


class RecordHeartbeat(Workflow):
    async def run(self, source: str) -> None:
        Heartbeat.record(source)

from druks.workflows import Workflow, step


class Sweep(Workflow):
    async def run_multistep(self, repo: str) -> None:
        findings = await self.scan(repo)
        await NightWatch.report(repo=repo, findings=findings)

    @step
    async def scan(self, repo: str) -> list[str]:
        return await scanner.scan(repo)

from druks.sandbox import Sandbox
from druks.workflows import Workflow


class BuildSite(Workflow):
    sandbox = Sandbox(setup="sandboxes/build.sh")

run_id = await Sweep.start(
    subject=repository,
    repo=full_name,
)

self.journal.filter(PlanData)                                # all entries, oldest first
self.journal.latest(PlanData)                                # newest, or None
self.journal.filter(ImplementationOutput, status="success")  # keyword filters: ANDed equality
self.journal.filter(ReviewWork)                              # gate replies, by their Gate class

class SweepJournal(Journal):
    @property
    def findings(self) -> list[FindingData]:
        return self.filter(FindingData)


class Sweep(Workflow):
    journal_class = SweepJournal

await self.announce("pr.opened", pr_number=delivery.pr_number, branch=delivery.branch)

class Sweep(Workflow):
    every = "0 6 * * *"

class Engage(Workflow):
    subject = Account
    every = "0 */4 * * *"

    @classmethod
    async def dispatch(cls) -> str:
        return await cls.start(subject=Account.get())

    async def run(self) -> None:
        ...

from druks.workflows import task


@task(every="*/15 * * * *")
async def refresh_tokens() -> None:
    ...


@task(retries=4)
async def sync_labels(pull_request_id: int) -> None:
    ...

from pydantic import BaseModel, Field


class Sweep(Workflow):
    class Settings(BaseModel):
        batch_size: int = Field(default=20, ge=1, le=100)

    @step
    async def load_settings(self) -> "Sweep.Settings":
        return self.settings()

from druks.agents import Agent, AgentOutput


class ReportOutput(AgentOutput):
    title: str
    body: str


class NightWatch(App):
    report = Agent(
        model="claude",
        prompt="night_watch/report.md",
        contract=ReportOutput,
        description="Turns findings into an operator report.",
    )

result = await NightWatch.report(repo=repo, findings=findings)

from druks.browser import BrowserSession
from druks.apps import App


class NightWatch(App):
    name = "night_watch"
    acme = BrowserSession(site="acme.example", persist=True)

async with NightWatch.acme.playwright() as browser:
    page = await browser.new_page()  # opened on the logged-in context
    await page.goto("https://acme.example/home")

from typing import Literal

from druks.workflows import Gate, Workflow


class ApproveReport(Gate):
    name = "approve_report"
    action: Literal["approve", "revise", "cancel"]
    note: str | None = None

    @classmethod
    async def on_wait(cls, workflow: Workflow) -> None:
        await notifier.report_ready(workflow.workflow_id)

reply = await ApproveReport.wait(
    input_request={
        "presentation": "external",
        "label": "Review the night-watch report",
        "url": review_url,
    }
)

await ApproveReport.answer(
    repository,
    action="approve",
    note="Ship it.",
)

reply = await self.review(questions=report.questions, context=review_context)

await Sweep.cancel(repository)

class Sweep(Workflow):
    subject = Repository

from druks.db import StoredSubject


class Repository(StoredSubject):
    __tablename__ = "repositories"

    def get_label(self) -> str:
        return self.full_name

    @classmethod
    def list_summaries(cls, account_id: str | None) -> list[SubjectSummary]:
        return [repository.get_summary() for repository in cls.list_open()]

from druks.workflows import SubjectSummary


class RepositorySummary(SubjectSummary):
    open_findings: int


class Repository(StoredSubject):
    def get_summary(self) -> RepositorySummary:
        return RepositorySummary.model_validate(self)

from druks.workflows import Subject


class PullRequest(Subject):
    @classmethod
    def list_summaries(cls, account_id: str | None) -> list[SubjectSummary]:
        return [pull_request.get_summary() for pull_request in cls.list_open()]

await NightWatch.dispatch(subject=repository)

status = repository.get_status()
if status.is_parked:
    ...  # a run stopped to ask a human something

NightWatch.record_event(
    type="report.published",
    subject=repository,
    payload={"url": report_url},
)

class Repository(StoredSubject):
    def get_label(self) -> str:
        return self.full_name

from druks.signals import subscribe
from druks.workflows import WorkflowEvent


@subscribe(WorkflowEvent.FINISHED, subject=Repository)
async def on_sweep_finished(*, subject: Repository, **_: object) -> None:
    await notify(subject.full_name)

from fastapi.responses import JSONResponse

from druks.signals import publish
from druks.webhooks import Webhook, verify_hmac_sha256


class NightWatchWebhook(Webhook):
    provider = "night_watch"
    category = "events"

    def request_is_authentic(self) -> bool:
        verify_hmac_sha256(
            self.raw_body,
            self.request.headers.get("x-signature"),
            secret,
        )
        return True

    def get_action(self) -> str:
        return self.data["type"].replace(".", "_")

    async def on_report_approved(self) -> JSONResponse:
        await publish("report.approved", payload=self.data)
        return JSONResponse({"accepted": True})

from sqlalchemy.orm import Mapped, mapped_column

from druks.db import Base


class Report(Base):
    __tablename__ = "night_watch_reports"

    id: Mapped[int] = mapped_column(primary_key=True)

uv run druks makemigrations night_watch -m "add reports"
uv run druks init-db

router = APIRouter(prefix="/reviews")

from druks.accounts import current_account_id

@router.get("/reviews")
def list_reviews() -> list[ReviewResponse]:
    return Review.list_for_account(current_account_id.get())

from pydantic import BaseModel, Field, SecretStr

from druks.services import Service, ServiceConnectError


class Gmail(Service):
    description = "The appliance's own OAuth client — every mailbox authenticates against it."

    class Settings(BaseModel):
        client_id: str = Field(title="Client ID")
        client_secret: SecretStr = Field(title="Client secret")

Gmail.get().secrets["client_secret"]   # raises ServiceNotConnectedError when unset
Gmail.is_connected()

@classmethod
async def verify(cls, settings: Settings) -> dict:
    if not await probe(settings):
        raise ServiceConnectError("The provider did not accept these credentials.")
    return {}

class Acme(Service):
    authorization_endpoint = "https://acme.example/oauth/authorize"
    token_endpoint = "https://acme.example/oauth/token"
    # True = HTTP Basic on the token endpoint. False = secret in the body.
    basic_auth = True
    identity_endpoint = "https://acme.example/oauth/userinfo"
    identity_scopes = ("openid", "email")
    # Query parameters the provider's consent URL must carry.
    extra_authorize_params = {"access_type": "offline", "prompt": "consent"}

    class Settings(BaseModel):
        client_id: str = Field(title="Client ID")
        client_secret: SecretStr = Field(title="Client secret")

    @classmethod
    async def get_identity(cls, access_token: str) -> dict:
        payload = await fetch_profile(access_token)
        return payload["data"]

class GoogleOauth(Service):
    abstract = True
    authorization_endpoint = "https://accounts.google.com/o/oauth2/v2/auth"
    token_endpoint = "https://oauth2.googleapis.com/token"
    extra_authorize_params = {"access_type": "offline", "prompt": "consent"}
    identity_endpoint = "https://openidconnect.googleapis.com/v1/userinfo"
    identity_scopes = ("openid", "email")
    identity_key = "sub"

    class Settings(BaseModel):
        client_id: str = Field(title="Client ID")
        client_secret: SecretStr = Field(title="Client secret")


class Gmail(GoogleOauth):
    pass


class GoogleCalendar(GoogleOauth):
    pass

class NightWatch(App):
    name = "night_watch"
    acme = Acme.with_scopes("profile.read", "posts.write")

for connection in NightWatch.acme.list_for_account(account_id):
    token = await connection.get_access_token()

from druks.signals import subscribe


@subscribe("oauth.connected", provider="acme")
async def adopt_sign_in(
    provider: str, connection_id: str, account_id: str, reconsent: bool
) -> None:
    if not reconsent:
        WatchedAccount.adopt(connection_id, account_id)


@subscribe("oauth.disconnected", provider="acme")
async def drop_sign_in(provider: str, connection_id: str, account_id: str) -> None:
    WatchedAccount.drop(connection_id)

from typing import Literal

from pydantic import Field

from druks.apps import App, AppSettings, Secret


class NightWatch(App):
    name = "night_watch"

    class Settings(AppSettings):
        provider: Literal["none", "acme"] = "none"
        service_token: Secret = Field(
            json_schema_extra={"section": "Acme", "visible_when": {"provider": "acme"}},
        )
        webhook_secret: Secret = Field(
            title="Webhook secret",
            json_schema_extra={"section": "Acme", "visible_when": {"provider": "acme"}},
        )

        def clean(self) -> dict[str, str]:
            if self.service_token and not self.webhook_secret:
                return {"webhook_secret": "Required once the service token is set."}
            return {}

from druks.testing import run_workflow

summary = await run_workflow(Sweep, subject=repository, since="2026-07-01")

from druks.testing import seed_call, seed_run

run = seed_run(
    druks_db,
    kind=Sweep.kind,
    subject=repository,
    state="running",
)
call = seed_call(
    druks_db,
    run,
    NightWatch.report,
    status="running",
)

class NightWatch(App):
    name = "night_watch"
    navigation = [("/night_watch", "reports")]