Loading...
Loading...
Create Galaxy REST API endpoints with FastAPI routers, Pydantic schemas, and manager pattern. Use for: new API routes, FastAPI endpoints, REST resources, Pydantic request/response models, lib/galaxy/webapps/galaxy/api routers, lib/galaxy/schema definitions, API controller creation.
npx skill4agent add arash77/galaxy-claude-marketplace galaxy-api-endpoint# Find recently modified API routers
ls -t lib/galaxy/webapps/galaxy/api/*.py | head -5lib/galaxy/webapps/galaxy/api/job_files.pylib/galaxy/webapps/galaxy/api/workflows.pylib/galaxy/webapps/galaxy/api/histories.pyrouter = APIRouter(tags=["resource_name"])DependsOnTransgalaxy.schemalib/galaxy/schema/lib/galaxy/schema/schema.pylib/galaxy/schema/workflows.pyfrom typing import Optional, List
from pydantic import Field
from galaxy.schema.fields import EncodedDatabaseIdField
from galaxy.schema.schema import Modelclass 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: intEncodedDatabaseIdFieldDecodedDatabaseIdFieldstrintboolfloatOptional[T]List[T]datetimedescription...Field(alias="...")lib/galaxy/managers/lib/galaxy/managers/<resource>s.pylib/galaxy/managers/<resource>s.pyfrom 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()apptransself.sa_sessiongalaxy.exceptionsselect()lib/galaxy/webapps/galaxy/api/lib/galaxy/webapps/galaxy/api/<resource>s.py"""
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"])
# Dependency for manager
def get_myresource_manager(app=Depends(get_app)) -> MyResourceManager:
return MyResourceManager(app)
@router.cbv
class FastAPIMyResources:
manager: MyResourceManager = Depends(get_myresource_manager)
@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,
)Routergalaxy.webapps.galaxy.api@router.cbvmanager: Manager = Depends(get_manager)DependsOnTransPath(...)Query(...)status_code=status.HTTP_201_CREATEDsummarylib/galaxy/webapps/galaxy/buildapp.pyfrom galaxy.webapps.galaxy.api import myresourcesapp_factory()app.include_router(myresources.router)app.include_router()lib/galaxy_test/api/lib/galaxy_test/api/test_<resource>s.py"""
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"]ApiTestCaselib/galaxy_test/api/_framework.pyself._get()self._post()self._put()self._delete()/api/self._assert_status_code_is(response, 200)self._assert_status_code_is_ok(response)self._assert_has_keys(obj, "key1", "key2")self._different_user()_create_myresource()# Run all tests for your new API
./run_tests.sh -api lib/galaxy_test/api/test_myresources.py
# Run a specific test
./run_tests.sh -api lib/galaxy_test/api/test_myresources.py::TestMyResourcesApi::test_create_myresource
# Run with verbose output
./run_tests.sh -api lib/galaxy_test/api/test_myresources.py --verbose_errors./run_tests.shpytest./run.shhttp://localhost:8080/api/docs# Create
curl -X POST http://localhost:8080/api/myresources \
-H "Content-Type: application/json" \
-d '{"name": "Test", "description": "Test resource"}'
# List
curl http://localhost:8080/api/myresources
# Get specific
curl http://localhost:8080/api/myresources/{id}client/src/api/schema/schema.tsls -t lib/galaxy/webapps/galaxy/api/*.py | head -5lib/galaxy/schema/schema.pylib/galaxy/schema/fields.pyls lib/galaxy/managers/*.pyls lib/galaxy_test/api/test_*.pytrans.security.encode_id()transself.sa_session.flush()commit()galaxy.exceptionsbuildapp.py./run_tests.sh -apipytest/api/docsreference.md