12222532506

Committed 08 Dec 2024 02:16PM UTC coverage: 97.373% (-0.7%) from 98.03%

Build # 12222532506

Build Type

Pull #39

github

Committed by

andersonresende

Commit Message

Layout adjustments

Pull Request Pull Request #39: Added item api

Run Details

78 of 82 branches covered (95.12%)

Branch coverage included in aggregate %.

169 of 184 new or added lines in 9 files covered. (91.85%)

16 existing lines in 1 file now uncovered.

1627 of 1669 relevant lines covered (97.48%)

6.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

61.9

/fastapi_backend/app/main.py

from uuid import UUID

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

from .database import User, get_async_session
from .models import Item
from .schemas import UserCreate, UserRead, UserUpdate, ItemRead, ItemCreate
from .users import auth_backend, current_active_user, fastapi_users, AUTH_URL_PATH
from fastapi.middleware.cors import CORSMiddleware
from .utils import simple_generate_unique_route_id
from sqlalchemy.future import select


app = FastAPI(generate_unique_id_function=simple_generate_unique_route_id)


origins = [
    "http://localhost:3000",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


app.include_router(
    fastapi_users.get_auth_router(auth_backend),
    prefix=f"/{AUTH_URL_PATH}/jwt",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_register_router(UserRead, UserCreate),
    prefix=f"/{AUTH_URL_PATH}",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_reset_password_router(),
    prefix=f"/{AUTH_URL_PATH}",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_verify_router(UserRead),
    prefix=f"/{AUTH_URL_PATH}",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_users_router(UserRead, UserUpdate),
    prefix="/users",
    tags=["users"],
)


@app.get("/authenticated-route", tags=["custom-auth"])
async def authenticated_route(user: User = Depends(current_active_user)):
    return {"message": f"Hello {user.email}!"}


@app.get("/items/", tags=["item"], response_model=list[ItemRead])
async def read_item(
    db: AsyncSession = Depends(get_async_session),
    user: User = Depends(current_active_user),
):
    result = await db.execute(select(Item).filter(Item.user_id == user.id))
    items = result.scalars().all()
    return [ItemRead.from_orm(item) for item in items]


@app.post("/items/", tags=["item"], response_model=ItemRead)
async def create_item(
    item: ItemCreate,
    db: AsyncSession = Depends(get_async_session),
    user: User = Depends(current_active_user),
):
    db_item = Item(**item.model_dump(), user_id=user.id)
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item


@app.delete("/items/{item_id}", tags=["item"])
async def delete_item(
    item_id: UUID,
    db: AsyncSession = Depends(get_async_session),
    user: User = Depends(current_active_user),
):
    result = await db.execute(
        select(Item).filter(Item.id == item_id, Item.user_id == user.id)
    )
    item = result.scalars().first()

    if not item:
        raise HTTPException(status_code=404, detail="Item not found or not authorized")

    await db.delete(item)
    await db.commit()

    return {"message": "Item successfully deleted"}

1	from uuid import UUID	1✔
2
3	from fastapi import Depends, FastAPI, HTTPException	1✔
4	from sqlalchemy.ext.asyncio import AsyncSession	1✔
5
6	from .database import User, get_async_session	1✔
7	from .models import Item	1✔
8	from .schemas import UserCreate, UserRead, UserUpdate, ItemRead, ItemCreate	1✔
9	from .users import auth_backend, current_active_user, fastapi_users, AUTH_URL_PATH	1✔
10	from fastapi.middleware.cors import CORSMiddleware	1✔
11	from .utils import simple_generate_unique_route_id	1✔
12	from sqlalchemy.future import select	1✔
13
14
15	app = FastAPI(generate_unique_id_function=simple_generate_unique_route_id)	1✔
16
17
18	origins = [	1✔
19	"http://localhost:3000",
20	"http://localhost:8080",
21	]
22
23	app.add_middleware(	1✔
24	CORSMiddleware,
25	allow_origins=origins,
26	allow_credentials=True,
27	allow_methods=["*"],
28	allow_headers=["*"],
29	)
30
31
32	app.include_router(	1✔
33	fastapi_users.get_auth_router(auth_backend),
34	prefix=f"/{AUTH_URL_PATH}/jwt",
35	tags=["auth"],
36	)
37	app.include_router(	1✔
38	fastapi_users.get_register_router(UserRead, UserCreate),
39	prefix=f"/{AUTH_URL_PATH}",
40	tags=["auth"],
41	)
42	app.include_router(	1✔
43	fastapi_users.get_reset_password_router(),
44	prefix=f"/{AUTH_URL_PATH}",
45	tags=["auth"],
46	)
47	app.include_router(	1✔
48	fastapi_users.get_verify_router(UserRead),
49	prefix=f"/{AUTH_URL_PATH}",
50	tags=["auth"],
51	)
52	app.include_router(	1✔
53	fastapi_users.get_users_router(UserRead, UserUpdate),
54	prefix="/users",
55	tags=["users"],
56	)
57
58
59	@app.get("/authenticated-route", tags=["custom-auth"])	1✔
60	async def authenticated_route(user: User = Depends(current_active_user)):	1✔
UNCOV 61	return {"message": f"Hello {user.email}!"}	×
62
63
64	@app.get("/items/", tags=["item"], response_model=list[ItemRead])	1✔
65	async def read_item(	1✔
66	db: AsyncSession = Depends(get_async_session),
67	user: User = Depends(current_active_user),
68	):
NEW UNCOV 69	result = await db.execute(select(Item).filter(Item.user_id == user.id))	×
NEW UNCOV 70	items = result.scalars().all()	×
NEW UNCOV 71	return [ItemRead.from_orm(item) for item in items]	×
72
73
74	@app.post("/items/", tags=["item"], response_model=ItemRead)	1✔
75	async def create_item(	1✔
76	item: ItemCreate,
77	db: AsyncSession = Depends(get_async_session),
78	user: User = Depends(current_active_user),
79	):
NEW UNCOV 80	db_item = Item(**item.model_dump(), user_id=user.id)	×
NEW UNCOV 81	db.add(db_item)	×
NEW UNCOV 82	await db.commit()	×
NEW UNCOV 83	await db.refresh(db_item)	×
NEW UNCOV 84	return db_item	×
85
86
87	@app.delete("/items/{item_id}", tags=["item"])	1✔
88	async def delete_item(	1✔
89	item_id: UUID,
90	db: AsyncSession = Depends(get_async_session),
91	user: User = Depends(current_active_user),
92	):
NEW UNCOV 93	result = await db.execute(	×
94	select(Item).filter(Item.id == item_id, Item.user_id == user.id)
95	)
NEW UNCOV 96	item = result.scalars().first()	×
97
NEW UNCOV 98	if not item:	×
NEW UNCOV 99	raise HTTPException(status_code=404, detail="Item not found or not authorized")	×
100
NEW UNCOV 101	await db.delete(item)	×
NEW UNCOV 102	await db.commit()	×
103
NEW UNCOV 104	return {"message": "Item successfully deleted"}	×

vintasoftware / nextjs-fastapi-template / 12222532506

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous