pyronaut create example.micronaut.micronautguide --features=data-jdbc,oracle,serialization-jackson

from dataclasses import dataclass
from typing import Annotated

from micronaut.data.annotation import GeneratedValue, Id, MappedEntity
from micronaut.serde.annotation import Serdeable


@Serdeable
@MappedEntity("TASK")
@dataclass
class Task:
    title: str
    completed: bool = False  (1)
    id: Annotated[int | None, Id, GeneratedValue] = None

CREATE TABLE "TASK" (
    "ID" NUMBER(19) NOT NULL PRIMARY KEY,
    "TITLE" VARCHAR(255) NOT NULL,
    "COMPLETED" BOOLEAN NOT NULL
)

from typing import Annotated

from micronaut.data.annotation import Id
from micronaut.data.jdbc.annotation import JdbcRepository
from micronaut.data.model.query.builder.sql import Dialect
from micronaut.data.repository import CrudRepository

from .task import Task


@JdbcRepository(dialect=Dialect.ORACLE, version="23.1")  (1)
class TaskRepository(CrudRepository[Task, int]):

    def findByCompletedTrue(self) -> list[Task]: ...  (2)

    def findByCompletedFalse(self) -> list[Task]: ...  (3)

    def findByCompleted(self, completed: bool) -> list[Task]: ...  (4)

    def updateCompleted(self, id: Annotated[int, Id], completed: bool) -> int: ...  (5)

from typing import Annotated

from micronaut.http import HttpResponse, HttpStatus
from micronaut.http.annotation import Body, Controller, Get, Post, Put, Status
from micronaut.scheduling import TaskExecutors
from micronaut.scheduling.annotation import ExecuteOn

from .task import Task
from .task_repository import TaskRepository


@ExecuteOn(TaskExecutors.BLOCKING)  (1)
@Controller("/tasks")
class TaskController:
    def __init__(self, task_repository: TaskRepository):
        self.task_repository = task_repository

    @Post
    @Status(HttpStatus.CREATED)
    def create(self, title: Annotated[str, Body("title")]) -> Task:  (2)
        return self.task_repository.save(Task(title))

    @Get("/open")
    def open(self) -> list[Task]:
        return self.task_repository.findByCompletedFalse()

    @Get("/completed")
    def completed(self) -> list[Task]:
        return self.task_repository.findByCompletedTrue()

    @Put("/{id}/complete")
    def complete(self, id: int) -> HttpResponse[Task]:
        if self.task_repository.updateCompleted(id, True) == 0:  (3)
            return HttpResponse.notFound()
        return HttpResponse.ok(self.task_repository.findById(id).orElseThrow())

import pytest
import requests
from micronaut.transaction import TransactionOperations
from pyronaut.test import MicronautTest, micronaut_test_fixture

from example.micronaut.task import Task
from example.micronaut.task_repository import TaskRepository


@pytest.fixture
def my_context(request):
    fixture = micronaut_test_fixture(
        request,
        MicronautTest(transactional=False),  (1)
    )
    yield fixture
    fixture.stop()


@pytest.fixture
def repository(my_context):
    repository = my_context[TaskRepository]
    repository.deleteAll()
    return repository


@pytest.fixture
def client(my_context):
    return requests.with_context(my_context)


def test_completed_column_is_native_boolean(my_context):
    assert column_data_type(my_context[TransactionOperations], "TASK", "COMPLETED") == "BOOLEAN"  (2)


def test_queries_boolean_values(repository):
    write = repository.save(Task("Write the guide"))
    review = repository.save(Task("Review the guide"))

    assert repository.updateCompleted(write.id, True) == 1  (3)

    assert titles(repository.findByCompletedTrue()) == ["Write the guide"]  (4)
    assert titles(repository.findByCompletedFalse()) == ["Review the guide"]
    assert titles(repository.findByCompleted(True)) == ["Write the guide"]  (5)

    assert repository.findById(write.id).orElseThrow().completed
    assert not repository.findById(review.id).orElseThrow().completed


def test_completes_task_over_http(repository, client):
    task = client.post("/tasks", json={"title": "Publish the guide"}).json()
    assert task["completed"] is False

    completed = client.put(f"/tasks/{task['id']}/complete").json()  (6)
    assert completed["completed"] is True

    assert [t["title"] for t in client.get("/tasks/completed").json()] == ["Publish the guide"]
    assert client.get("/tasks/open").json() == []


def titles(tasks) -> list[str]:
    return [task.title for task in tasks]


def column_data_type(transaction_operations: TransactionOperations, table: str, column: str) -> str | None:
    def read(status):
        sql = "SELECT data_type FROM user_tab_columns WHERE table_name = ? AND column_name = ?"
        statement = status.getConnection().prepareStatement(sql)
        try:
            statement.setString(1, table)
            statement.setString(2, column)
            result = statement.executeQuery()
            return result.getString(1) if result.next() else None
        finally:
            statement.close()

    return transaction_operations.executeRead(read)

pyronaut install
pyronaut validate-config
pyronaut test

curl -X POST http://localhost:8080/tasks \
     -H 'Content-Type: application/json' \
     -d '{"title":"Write the guide"}'

curl -X POST http://localhost:8080/tasks \
     -H 'Content-Type: application/json' \
     -d '{"title":"Review the guide"}'

curl -X PUT http://localhost:8080/tasks/1/complete

curl http://localhost:8080/tasks/completed

curl http://localhost:8080/tasks/open