release

src/app/api/api_v1/endpoints/items.py

@@ -12,11 +12,21 @@ async def create_item(item_in: ItemCreate, session: SessionDep) -> Item:
     return await item.create(session, obj_in=item_in)
 
 
-@router.get("/read-item", response_model=list[Item])
+@router.get("/read-all-item", response_model=list[Item])
 async def read_items(session: SessionDep) -> list[Item]:
     return await item.get_all(session)
 
 
+@router.get("/get-by-id", response_model=Item)
+async def read_item_by_id(id: str, session: SessionDep) -> Item | None:
+    return await item.get(session, id=id)
+
+
+@router.get("/get-by-owner", response_model=list[Item])
+async def read_item_by_owner(session: SessionDep) -> list[Item]:
+    return await item.get_multi_by_owner(session)
+
+
 @router.put("/update-item", response_model=Item)
 async def update_item(item_in: ItemUpdate, session: SessionDep) -> Item:
     return await item.update(session, obj_in=item_in)

src/app/api/deps.py

@@ -8,30 +8,42 @@
 import logging
 from typing import Annotated
 
-from fastapi import Depends, HTTPException, status
+from fastapi import Depends, HTTPException
 from fastapi.security import OAuth2PasswordBearer
 from gotrue import User
 from supabase_py_async import AsyncClient, create_client
 from supabase_py_async.lib.client_options import ClientOptions
 
 from app.core.config import settings
-from app.core.events import super_client
 
+
+async def supser_client() -> AsyncClient:
+    client: AsyncClient | None = None
+    try:
+        client = await create_client(
+            settings.SUPABASE_URL,
+            settings.SUPABASE_KEY,
+            options=ClientOptions(
+                postgrest_client_timeout=10, storage_client_timeout=10
+            ),
+        )
+        yield client
+    finally:
+        if client:
+            await client.auth.sign_out()
+
+
+SuperClientDep = Annotated[AsyncClient, Depends(supser_client)]
+
+# auto get access_token from header
 reusable_oauth2 = OAuth2PasswordBearer(
     tokenUrl="please login by supabase-js to get token"
 )
+
 TokenDep = Annotated[str, Depends(reusable_oauth2)]
 
 
-async def validate_user(token: str = Depends(reusable_oauth2)) -> str:
-    prefix = "Bearer "
-    if not token.startswith(prefix):
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid authentication credentials",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-    access_token = token[len(prefix) :]
+async def validate_user(access_token: TokenDep, super_client: SuperClientDep) -> str:
     try:
         if not super_client:
             raise HTTPException(status_code=401, detail="No super client")
@@ -55,6 +67,8 @@ async def get_db(access_token: AccessTokenDep) -> AsyncClient:
                 postgrest_client_timeout=10, storage_client_timeout=10
             ),
         )
+        # client.postgrest.auth(token=access_token)
+        # await client.auth.get_user(access_token)
         yield client
     except Exception as e:
         logging.error(e)

src/app/core/events.py

@@ -6,30 +6,12 @@
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
-from supabase_py_async import AsyncClient, create_client
-from supabase_py_async.lib.client_options import ClientOptions
-
-from app.core.config import settings
-
-super_client: AsyncClient | None = None
-
-
-async def create_super_client() -> AsyncClient:
-    return await create_client(
-        settings.SUPABASE_URL,
-        settings.SUPABASE_KEY,
-        options=ClientOptions(postgrest_client_timeout=10, storage_client_timeout=10),
-    )
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """life span events"""
-    global super_client
     try:
-        # start client
-        super_client = await create_super_client()
-
         yield
     finally:
         logging.info("lifespan shutdown")

tests/.env

@@ -0,0 +1,2 @@
+SUPABASE_TEST_URL=https://vtqsboqphlisizfwhrtg.supabase.co
+SUPABASE_TEST_KEY=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6InZ0cXNib3FwaGxpc2l6ZndocnRnIiwicm9sZSI6ImFub24iLCJpYXQiOjE3MDQ1MzY4NzIsImV4cCI6MjAyMDExMjg3Mn0.OgApaTXDr7llopKTplpXCsUzDubjbiQFXlaaFf6wlbY

tests/api/api_v1/test_items.py

@@ -1,37 +1,43 @@
-# from fastapi.testclient import TestClient
+# Additional assertions based on your application's logic
+
+# def test_create_item(test_client: TestClient,access_token: str):
+#     test_data = Faker().sentence()
+#     headers = get_auth_header(access_token)
+#     response = test_client.post("/create-item",headers=headers,
+#     json={"test_data": test_data})
+#     assert response.status_code == 200
+#     assert response.json()["test_data"] == "Sample Data"
+#     # Additional assertions based on your application's logic
 #
-# from src.app.core.config import settings
-# from tests.utils.item import create_random_item
+# def test_read_all_items(test_client):
+#     response = test_client.get("/read-all-item")
+#     assert response.status_code == 200
+#     assert isinstance(response.json(), list)
+#     # Further assertions can be added here
 #
+# def test_read_item_by_id(test_client):
+#     # Use a valid ID for an existing item
+#     response = test_client.get("/get-by-id?id=valid_id")
+#     assert response.status_code == 200
+#     assert response.json()["id"] == "valid_id"
+#     # More assertions based on expected item details
 #
-# def test_create_item(
-#     client: TestClient, superuser_token_headers: dict, db: Session
-# ) -> None:
-#     data = {"title": "Foo", "description": "Fighters"}
-#     response = client.post(
-#         f"{settings.API_V1_STR}/items/",
-#         headers=superuser_token_headers,
-#         json=data,
-#     )
+# def test_read_item_by_owner(test_client):
+#     response = test_client.get("/get-by-owner")
 #     assert response.status_code == 200
-#     content = response.json()
-#     assert content["title"] == data["title"]
-#     assert content["description"] == data["description"]
-#     assert "id" in content
-#     assert "owner_id" in content
+#     assert isinstance(response.json(), list)
+#     # Additional checks based on expected output
 #
+# def test_update_item(test_client):
+#     # Use a valid ID for an existing item
+#     response = test_client.put("/update-it
+#     em", json={"id": "valid_id", "test_data": "Updated Data"})
+#     assert response.status_code == 200
+#     assert response.json()["test_data"] == "Updated Data"
+#     # Additional checks and validations
 #
-# def test_read_item(
-#     client: TestClient, superuser_token_headers: dict, db: Session
-# ) -> None:
-#     item = create_random_item(db)
-#     response = client.get(
-#         f"{settings.API_V1_STR}/items/{item.id}",
-#         headers=superuser_token_headers,
-#     )
+# def test_delete_item(test_client):
+#     # Use a valid ID for an existing item
+#     response = test_client.delete("/delete", json={"id": "valid_id"})
 #     assert response.status_code == 200
-#     content = response.json()
-#     assert content["title"] == item.title
-#     assert content["description"] == item.description
-#     assert content["id"] == item.id
-#     assert content["owner_id"] == item.owner_id
+#     # Assertions to confirm deletion

tests/conftest.py

@@ -6,6 +6,7 @@
 
 import pytest
 from dotenv import load_dotenv
+from faker import Faker
 from fastapi.testclient import TestClient
 from pydantic import ConfigDict
 from supabase_py_async import AsyncClient, create_client
@@ -55,11 +56,38 @@ def pytest_configure(config: ConfigDict) -> None:
 
 
 @pytest.fixture(scope="module")
-def client() -> Generator:
+def client() -> Generator[TestClient, None, None]:
     with TestClient(app) as c:
         yield c
 
 
+@pytest.fixture(scope="module")
+async def access_token() -> AsyncGenerator[str, None]:
+    url = os.environ.get("SUPABASE_TEST_URL")
+    assert url is not None, "Must provide SUPABASE_TEST_URL environment variable"
+    key = os.environ.get("SUPABASE_TEST_KEY")
+    assert key is not None, "Must provide SUPABASE_TEST_KEY environment variable"
+    db_client = await create_client(url, key)
+    fake_email = Faker().email()
+    fake_password = Faker().password()
+    await db_client.auth.sign_up({"email": fake_email, "password": fake_password})
+    response = await db_client.auth.sign_in_with_password(
+        {"email": fake_email, "password": fake_password}
+    )
+    assert response.user.email == fake_email
+    assert response.user.id is not None
+    assert response.session.access_token is not None
+    assert response.session.refresh_token is not None
+    try:
+        yield response.session.access_token
+    finally:
+        await db_client.auth.sign_in_with_password(
+            {"email": "[email protected]", "password": "Zz030327#"}
+        )
+        await db_client.table("users").delete().eq("id", response.user.id).execute()
+        await db_client.auth.sign_out()
+
+
 @pytest.fixture(scope="module")
 async def db() -> AsyncGenerator[AsyncClient, None]:
     url = os.environ.get("SUPABASE_TEST_URL")
@@ -78,3 +106,17 @@ async def db() -> AsyncGenerator[AsyncClient, None]:
     finally:
         if db_client:
             await db_client.auth.sign_out()
+
+
+# FIXME: failed to auth
+# @pytest.mark.anyio
+# async def test_create_item(client: TestClient, access_token: str):
+#     from tests.utils import get_auth_header
+#     test_data = Faker().sentence()
+#     assert isinstance(access_token, str)
+#     headers = get_auth_header(access_token)
+#
+#     response = client.post("/api/v1/i
+#     tems/create-item", headers=headers, json={"test_data": test_data})
+#     assert response.status_code == 200
+#     assert response.json()["test_data"] == "Sample Data"

tests/utils.py

@@ -0,0 +1,11 @@
+"""
+-*- coding: utf-8 -*-
+@Organization : SupaVision
+@Author       : 18317
+@Date Created : 12/01/2024
+@Description  :
+"""
+
+
+def get_auth_header(access_token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {access_token}"}