8169178037

Committed 06 Mar 2024 08:39AM UTC coverage: 0.106%. First build

Build # 8169178037

Build Type

Pull #79

github

Committed by

henrikstranneheim

Commit Message

Merge branch 'main' into add-black

# Conflicts:
#	requirements.txt

Pull Request Pull Request #79: feat(black)

Run Details

1 of 940 relevant lines covered (0.11%)

0.0 hits per line

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

0.0

/genotype_api/api/endpoints/users.py

"""Routes for users"""

from typing import List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import EmailStr
from starlette import status
from starlette.responses import JSONResponse

from genotype_api.crud.users import get_user
from genotype_api.database import get_session
from genotype_api.models import User, UserRead, UserCreate, UserReadWithPlates
from sqlmodel import Session, select

from genotype_api.security import get_active_user
from sqlmodel.sql.expression import Select, SelectOfScalar

SelectOfScalar.inherit_cache = True
Select.inherit_cache = True


router = APIRouter()


@router.get("/{user_id}", response_model=UserReadWithPlates)
def read_user(
    user_id: int,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
) -> User:
    return get_user(session=session, user_id=user_id)


@router.delete("/{user_id}")
def delete_user(
    user_id: int,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
) -> JSONResponse:
    user: User = get_user(session=session, user_id=user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    if user.plates:
        raise HTTPException(
            status_code=status.HTTP_406_NOT_ACCEPTABLE,
            detail="User previously signed plates, please archive instead",
        )
    session.delete(user)
    session.commit()
    return JSONResponse(content="User deleted successfully", status_code=status.HTTP_200_OK)


@router.put("/{user_id}/email", response_model=User)
def change_user_email(
    user_id: int,
    email: EmailStr,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
) -> User:
    user: User = get_user(session=session, user_id=user_id)
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    user.email = email
    session.add(user)
    session.commit()
    session.refresh(user)
    return user


@router.get("/", response_model=List[UserRead])
def read_users(
    skip: int = 0,
    limit: int = Query(default=100, lte=100),
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
) -> List[User]:
    users: List[User] = session.exec(select(User).offset(skip).limit(limit)).all()
    return users


@router.post("/", response_model=User)
def create_user(
    user: UserCreate,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_active_user),
):
    user_in_db: List[User] = session.exec(select(User).where(User.email == user.email)).all()
    if user_in_db:
        raise HTTPException(status_code=409, detail="Email already registered")
    db_user = User.from_orm(user)
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

1	"""Routes for users"""
2
3	from typing import List, Optional	×
4
5	from fastapi import APIRouter, Depends, HTTPException, Query	×
6	from pydantic import EmailStr	×
7	from starlette import status	×
8	from starlette.responses import JSONResponse	×
9
10	from genotype_api.crud.users import get_user	×
11	from genotype_api.database import get_session	×
12	from genotype_api.models import User, UserRead, UserCreate, UserReadWithPlates	×
13	from sqlmodel import Session, select	×
14
15	from genotype_api.security import get_active_user	×
16	from sqlmodel.sql.expression import Select, SelectOfScalar	×
17
18	SelectOfScalar.inherit_cache = True	×
19	Select.inherit_cache = True	×
20
21
22	router = APIRouter()	×
23
24
25	@router.get("/{user_id}", response_model=UserReadWithPlates)	×
26	def read_user(	×
27	user_id: int,
28	session: Session = Depends(get_session),
29	current_user: User = Depends(get_active_user),
30	) -> User:
31	return get_user(session=session, user_id=user_id)	×
32
33
34	@router.delete("/{user_id}")	×
35	def delete_user(	×
36	user_id: int,
37	session: Session = Depends(get_session),
38	current_user: User = Depends(get_active_user),
39	) -> JSONResponse:
40	user: User = get_user(session=session, user_id=user_id)	×
41	if not user:	×
42	raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")	×
43	if user.plates:	×
44	raise HTTPException(	×
45	status_code=status.HTTP_406_NOT_ACCEPTABLE,
46	detail="User previously signed plates, please archive instead",
47	)
48	session.delete(user)	×
49	session.commit()	×
50	return JSONResponse(content="User deleted successfully", status_code=status.HTTP_200_OK)	×
51
52
53	@router.put("/{user_id}/email", response_model=User)	×
54	def change_user_email(	×
55	user_id: int,
56	email: EmailStr,
57	session: Session = Depends(get_session),
58	current_user: User = Depends(get_active_user),
59	) -> User:
60	user: User = get_user(session=session, user_id=user_id)	×
61	if not user:	×
62	raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")	×
63	user.email = email	×
64	session.add(user)	×
65	session.commit()	×
66	session.refresh(user)	×
67	return user	×
68
69
70	@router.get("/", response_model=List[UserRead])	×
71	def read_users(	×
72	skip: int = 0,
73	limit: int = Query(default=100, lte=100),
74	session: Session = Depends(get_session),
75	current_user: User = Depends(get_active_user),
76	) -> List[User]:
77	users: List[User] = session.exec(select(User).offset(skip).limit(limit)).all()	×
78	return users	×
79
80
81	@router.post("/", response_model=User)	×
82	def create_user(	×
83	user: UserCreate,
84	session: Session = Depends(get_session),
85	current_user: User = Depends(get_active_user),
86	):
87	user_in_db: List[User] = session.exec(select(User).where(User.email == user.email)).all()	×
88	if user_in_db:	×
89	raise HTTPException(status_code=409, detail="Email already registered")	×
90	db_user = User.from_orm(user)	×
91	session.add(db_user)	×
92	session.commit()	×
93	session.refresh(db_user)	×
94	return db_user	×

Clinical-Genomics / genotype-api / 8169178037

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