undefined

Read one of these files to understand current patterns. Good examples:
- `lib/galaxy/webapps/galaxy/api/job_files.py` - Simple CRUD operations
- `lib/galaxy/webapps/galaxy/api/workflows.py` - Complex resource with many operations
- `lib/galaxy/webapps/galaxy/api/histories.py` - RESTful resource with nested routes

**Key patterns to observe:**
- Router setup: `router = APIRouter(tags=["resource_name"])`
- Dependency injection: `DependsOnTrans`, custom dependency functions
- Request/response models: Pydantic schemas from `galaxy.schema`
- Error handling: Raising appropriate HTTP exceptions
- Documentation: Docstrings and OpenAPI metadata

---

from typing import Optional, List
from pydantic import Field
from galaxy.schema.fields import EncodedDatabaseIdField
from galaxy.schema.schema import Model

class MyResourceCreateRequest(Model):
    """Request model for creating a new resource."""
    name: str = Field(..., description="Resource name")
    description: Optional[str] = Field(None, description="Optional description")

class MyResourceResponse(Model):
    """Response model for resource operations."""
    id: EncodedDatabaseIdField = Field(..., description="Encoded resource ID")
    name: str
    description: Optional[str]
    create_time: datetime
    update_time: datetime

class MyResourceListResponse(Model):
    """Response model for listing resources."""
    items: List[MyResourceResponse]
    total_count: int

from typing import Optional
from galaxy import model
from galaxy.managers.context import ProvidesUserContext
from galaxy.model import Session

class MyResourceManager:
    """Manager for MyResource operations."""

    def __init__(self, app):
        self.app = app
        self.sa_session: Session = app.model.context

    def create(
        self,
        trans: ProvidesUserContext,
        name: str,
        description: Optional[str] = None
    ) -> model.MyResource:
        """Create a new resource."""
        resource = model.MyResource(
            user=trans.user,
            name=name,
            description=description
        )
        self.sa_session.add(resource)
        self.sa_session.flush()
        return resource

    def get(self, trans: ProvidesUserContext, resource_id: int) -> model.MyResource:
        """Get resource by ID."""
        resource = self.sa_session.get(model.MyResource, resource_id)
        if not resource:
            raise exceptions.ObjectNotFound("Resource not found")
        if not self.is_accessible(resource, trans.user):
            raise exceptions.ItemAccessibilityException("Access denied")
        return resource

    def is_accessible(self, resource: model.MyResource, user: Optional[model.User]) -> bool:
        """Check if user can access this resource."""
        if not user:
            return False
        return resource.user_id == user.id

    def list_for_user(self, trans: ProvidesUserContext) -> List[model.MyResource]:
        """List all resources for the current user."""
        stmt = select(model.MyResource).where(
            model.MyResource.user_id == trans.user.id
        )
        return self.sa_session.scalars(stmt).all()

"""
API endpoints for MyResource operations.
"""
import logging
from typing import Optional

from fastapi import (
    APIRouter,
    Depends,
    Path,
    Query,
    status,
)

from galaxy.managers.context import ProvidesUserContext
from galaxy.managers.myresources import MyResourceManager
from galaxy.schema.schema import (
    MyResourceCreateRequest,
    MyResourceResponse,
    MyResourceListResponse,
)
from galaxy.webapps.galaxy.api import (
    DependsOnTrans,
    Router,
)
from galaxy.webapps.galaxy.api.depends import get_app

log = logging.getLogger(__name__)

router = Router(tags=["myresources"])

@router.get(
    "/api/myresources",
    summary="List all resources for current user",
    response_model=MyResourceListResponse,
)
def index(
    self,
    trans: ProvidesUserContext = DependsOnTrans,
) -> MyResourceListResponse:
    """List all resources owned by the current user."""
    items = self.manager.list_for_user(trans)
    return MyResourceListResponse(
        items=[self._serialize(item) for item in items],
        total_count=len(items),
    )

@router.post(
    "/api/myresources",
    summary="Create a new resource",
    status_code=status.HTTP_201_CREATED,
    response_model=MyResourceResponse,
)
def create(
    self,
    trans: ProvidesUserContext = DependsOnTrans,
    request: MyResourceCreateRequest = ...,
) -> MyResourceResponse:
    """Create a new resource."""
    resource = self.manager.create(
        trans,
        name=request.name,
        description=request.description,
    )
    return self._serialize(resource)

@router.get(
    "/api/myresources/{id}",
    summary="Get resource by ID",
    response_model=MyResourceResponse,
)
def show(
    self,
    trans: ProvidesUserContext = DependsOnTrans,
    id: EncodedDatabaseIdField = Path(..., description="Resource ID"),
) -> MyResourceResponse:
    """Get a specific resource by ID."""
    decoded_id = trans.security.decode_id(id)
    resource = self.manager.get(trans, decoded_id)
    return self._serialize(resource)

def _serialize(self, resource) -> MyResourceResponse:
    """Convert model object to response schema."""
    return MyResourceResponse(
        id=trans.security.encode_id(resource.id),
        name=resource.name,
        description=resource.description,
        create_time=resource.create_time,
        update_time=resource.update_time,
    )

**Router best practices:**
- Use `Router` (capital R) from `galaxy.webapps.galaxy.api` (subclass of FastAPI's APIRouter)
- Use `@router.cbv` class-based views for grouping related endpoints
- Use dependency injection for managers: `manager: Manager = Depends(get_manager)`
- Use `DependsOnTrans` for transaction context
- Path parameters use `Path(...)` with descriptions
- Query parameters use `Query(...)` with defaults
- Set appropriate HTTP status codes (`status_code=status.HTTP_201_CREATED` for creates)
- Add `summary` to all endpoints for OpenAPI docs
- Decode IDs in endpoint, not in manager (manager works with integer IDs)

---

from galaxy.webapps.galaxy.api import myresources

app.include_router(myresources.router)

"""
API tests for MyResource endpoints.
"""
from galaxy_test.base.populators import DatasetPopulator
from ._framework import ApiTestCase


class TestMyResourcesApi(ApiTestCase):
    """Tests for /api/myresources endpoints."""

    def setUp(self):
        super().setUp()
        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)

    def test_create_myresource(self):
        """Test creating a new resource."""
        payload = {
            "name": "Test Resource",
            "description": "Test description",
        }
        response = self._post("myresources", data=payload, json=True)
        self._assert_status_code_is(response, 201)
        resource = response.json()
        self._assert_has_keys(resource, "id", "name", "description", "create_time")
        assert resource["name"] == "Test Resource"

    def test_list_myresources(self):
        """Test listing resources."""
        # Create some test data
        self._create_myresource("Resource 1")
        self._create_myresource("Resource 2")

        # List resources
        response = self._get("myresources")
        self._assert_status_code_is_ok(response)
        data = response.json()
        assert data["total_count"] >= 2
        assert len(data["items"]) >= 2

    def test_get_myresource(self):
        """Test getting a specific resource."""
        resource_id = self._create_myresource("Test Resource")
        response = self._get(f"myresources/{resource_id}")
        self._assert_status_code_is_ok(response)
        resource = response.json()
        assert resource["id"] == resource_id
        assert resource["name"] == "Test Resource"

    def test_get_nonexistent_myresource(self):
        """Test getting a resource that doesn't exist."""
        response = self._get("myresources/invalid_id")
        self._assert_status_code_is(response, 404)

    def test_create_myresource_as_different_user(self):
        """Test that users can only see their own resources."""
        # Create as first user
        resource_id = self._create_myresource("User 1 Resource")

        # Switch to different user
        with self._different_user():
            # Should not be able to access
            response = self._get(f"myresources/{resource_id}")
            self._assert_status_code_is(response, 403)

    def _create_myresource(self, name: str) -> str:
        """Helper to create a resource and return its ID."""
        payload = {"name": name, "description": f"Description for {name}"}
        response = self._post("myresources", data=payload, json=True)
        self._assert_status_code_is(response, 201)
        return response.json()["id"]

undefined

**IMPORTANT:** Always use `./run_tests.sh`, not `pytest` directly. The wrapper script sets up the correct environment.

---

ls -t lib/galaxy/webapps/galaxy/api/*.py | head -5

ls lib/galaxy/managers/*.py

ls lib/galaxy_test/api/test_*.py