Implemented GET/deployed-contracts-endpoints

py-be/app/api/routes.py

@@ -3,11 +3,12 @@
 from datetime import datetime
 
 from fastapi import Depends, FastAPI, HTTPException, status
-from pydantic import BaseModel, ConfigDict, constr, field_validator
+from pydantic import BaseModel, ConfigDict, Field, constr, field_validator
 from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
 from ..models.base import init_db
+from ..models.deployed_contract import DeployedContract
 from ..models.generated_contract import GeneratedContract
 from ..models.user import User
 from ..services.base import get_db
@@ -92,6 +93,18 @@ class GeneratedContractRead(BaseModel):
     updated_at: datetime
 
 
+class DeployedContractRead(BaseModel):
+    """Schema returned for deployed contracts."""
+
+    model_config = ConfigDict(from_attributes=True, populate_by_name=True)
+
+    id: int
+    contract_name: str
+    contract_address: str
+    metadata: dict | None = Field(None, alias="contract_metadata")
+    deployed_at: datetime
+
+
 @app.post(
     "/generate",
     response_model=GeneratedContractRead,
@@ -149,3 +162,39 @@ def generate_contract(
     db.commit()
     db.refresh(contract)
     return contract
+
+
+@app.get(
+    "/deployed_contracts",
+    response_model=list[DeployedContractRead],
+    status_code=status.HTTP_200_OK,
+)
+def get_deployed_contracts(
+    name: str | None = None,
+    sort_by: str = "deployed_at",
+    order: str = "desc",
+    db: Session = Depends(get_db),
+) -> list[DeployedContract]:
+    """Retrieve deployed contracts with optional filtering and sorting."""
+
+    valid_sort = {
+        "deployed_at": DeployedContract.deployed_at,
+        "contract_name": DeployedContract.contract_name,
+    }
+    if sort_by not in valid_sort:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid sort_by field",
+        )
+
+    query = db.query(DeployedContract)
+    if name:
+        query = query.filter(DeployedContract.contract_name.ilike(f"%{name}%"))
+
+    sort_column = valid_sort[sort_by]
+    if order == "desc":
+        query = query.order_by(sort_column.desc())
+    else:
+        query = query.order_by(sort_column.asc())
+
+    return query.all()

py-be/app/models/base.py

@@ -19,6 +19,6 @@
 def init_db() -> None:
     """Create database tables."""
     # Import models here to ensure they are registered with SQLAlchemy
-    from . import generated_contract, user  # noqa: F401
+    from . import deployed_contract, generated_contract, user  # noqa: F401
 
     Base.metadata.create_all(bind=engine)

py-be/app/models/deployed_contracts.py

@@ -0,0 +1,19 @@
+"""Model for storing deployed contract metadata."""
+
+from datetime import datetime
+
+from sqlalchemy import JSON, Column, DateTime, Integer, String
+
+from .base import Base
+
+
+class DeployedContract(Base):
+    """SQLAlchemy model for a deployed contract."""
+
+    __tablename__ = "deployed_contracts"
+
+    id = Column(Integer, primary_key=True, index=True)
+    contract_name = Column(String, nullable=False)
+    contract_address = Column(String, unique=True, nullable=False, index=True)
+    contract_metadata = Column("metadata", JSON, nullable=True)
+    deployed_at = Column(DateTime, default=datetime.utcnow)

py-be/tests/conftest.py

@@ -8,13 +8,17 @@
 from app.models.base import Base
 from app.services.base import get_db
 
+TEST_DATABASE_URL = os.getenv("TEST_DATABASE_URL", "sqlite:///./test.db")
+
 TEST_DATABASE_URL = os.getenv(
     "TEST_DATABASE_URL",
     "postgresql://postgres:Soham2003@localhost:5432/starkfinder_test",
 )
+os.environ["DATABASE_URL"] = TEST_DATABASE_URL
 
 
 engine = create_engine(TEST_DATABASE_URL)
+engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
 TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
 
@@ -29,6 +33,12 @@ def setup_test_database():
     yield
 
     drop_database(TEST_DATABASE_URL)
+    Base.metadata.drop_all(bind=engine)
+    if TEST_DATABASE_URL.startswith("sqlite"):
+        try:
+            os.remove("test.db")
+        except FileNotFoundError:
+            pass
 
 
 @pytest.fixture()

py-be/tests/test_deployed_contracts.py

@@ -0,0 +1,57 @@
+"""Tests for the GET /deployed_contracts endpoint."""
+
+from datetime import datetime, timedelta
+
+from fastapi.testclient import TestClient
+
+from app.api import routes
+from app.models.deployed_contract import DeployedContract
+
+client = TestClient(routes.app)
+
+
+def seed_contracts(db):
+    """Insert sample deployed contracts into the database."""
+    db.query(DeployedContract).delete()
+    now = datetime.utcnow()
+    contracts = [
+        DeployedContract(
+            contract_name="A",
+            contract_address="0x1",
+            contract_metadata={"v": 1},
+            deployed_at=now - timedelta(days=1),
+        ),
+        DeployedContract(
+            contract_name="B",
+            contract_address="0x2",
+            deployed_at=now,
+        ),
+        DeployedContract(
+            contract_name="C",
+            contract_address="0x3",
+            deployed_at=now - timedelta(days=2),
+        ),
+    ]
+    db.add_all(contracts)
+    db.commit()
+    return contracts
+
+
+def test_list_deployed_contracts_default_order(db_session):
+    seed_contracts(db_session)
+    res = client.get("/deployed_contracts")
+    assert res.status_code == 200
+    data = res.json()
+    assert [c["contract_name"] for c in data] == ["B", "A", "C"]
+
+
+def test_list_deployed_contracts_filter_and_sort(db_session):
+    seed_contracts(db_session)
+    res = client.get(
+        "/deployed_contracts",
+        params={"name": "A", "sort_by": "contract_name", "order": "asc"},
+    )
+    assert res.status_code == 200
+    data = res.json()
+    assert len(data) == 1
+    assert data[0]["contract_name"] == "A"