Remove some old crud functions

This commit is contained in:
Sebastiaan
2025-05-24 11:04:23 +02:00
parent c1442f430e
commit 28eb10fcce
12 changed files with 135 additions and 126 deletions

View File

@@ -5,12 +5,12 @@ from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import HTMLResponse from fastapi.responses import HTMLResponse
from fastapi.security import OAuth2PasswordRequestForm from fastapi.security import OAuth2PasswordRequestForm
from app import crud
from app.api.deps import CurrentUser, SessionDep, get_current_active_superuser from app.api.deps import CurrentUser, SessionDep, get_current_active_superuser
from app.core import security from app.core import security
from app.core.config import settings from app.core.config import settings
from app.core.security import get_password_hash from app.core.security import get_password_hash
from app.models import Message, NewPassword, Token, UserPublic from app.models.base import Message
from app.models.user import User, NewPassword, Token, UserPublic
from app.utils import ( from app.utils import (
generate_password_reset_token, generate_password_reset_token,
generate_reset_password_email, generate_reset_password_email,
@@ -28,7 +28,7 @@ def login_access_token(
""" """
OAuth2 compatible token login, get an access token for future requests OAuth2 compatible token login, get an access token for future requests
""" """
user = crud.authenticate( user = User.authenticate(
session=session, email=form_data.username, password=form_data.password session=session, email=form_data.username, password=form_data.password
) )
if not user: if not user:
@@ -56,7 +56,7 @@ def recover_password(email: str, session: SessionDep) -> Message:
""" """
Password Recovery Password Recovery
""" """
user = crud.get_user_by_email(session=session, email=email) user = User.get_by_email(session=session, email=email)
if not user: if not user:
raise HTTPException( raise HTTPException(
@@ -83,7 +83,7 @@ def reset_password(session: SessionDep, body: NewPassword) -> Message:
email = verify_password_reset_token(token=body.token) email = verify_password_reset_token(token=body.token)
if not email: if not email:
raise HTTPException(status_code=400, detail="Invalid token") raise HTTPException(status_code=400, detail="Invalid token")
user = crud.get_user_by_email(session=session, email=email) user = User.get_by_email(session=session, email=email)
if not user: if not user:
raise HTTPException( raise HTTPException(
status_code=404, status_code=404,
@@ -107,7 +107,7 @@ def recover_password_html_content(email: str, session: SessionDep) -> Any:
""" """
HTML Content for Password Recovery HTML Content for Password Recovery
""" """
user = crud.get_user_by_email(session=session, email=email) user = User.get_by_email(session=session, email=email)
if not user: if not user:
raise HTTPException( raise HTTPException(

View File

@@ -4,7 +4,6 @@ from typing import Any
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import col, delete, func, select from sqlmodel import col, delete, func, select
from app import crud
from app.api.deps import ( from app.api.deps import (
CurrentUser, CurrentUser,
SessionDep, SessionDep,
@@ -12,9 +11,8 @@ from app.api.deps import (
) )
from app.core.config import settings from app.core.config import settings
from app.core.security import get_password_hash, verify_password from app.core.security import get_password_hash, verify_password
from app.models import ( from app.models.base import Message
Item, from app.models.user import (
Message,
UpdatePassword, UpdatePassword,
User, User,
UserCreate, UserCreate,
@@ -55,14 +53,14 @@ def create_user(*, session: SessionDep, user_in: UserCreate) -> Any:
""" """
Create new user. Create new user.
""" """
user = crud.get_user_by_email(session=session, email=user_in.email) user = User.get_by_email(session=session, email=user_in.email)
if user: if user:
raise HTTPException( raise HTTPException(
status_code=400, status_code=400,
detail="The user with this email already exists in the system.", detail="The user with this email already exists in the system.",
) )
user = crud.create_user(session=session, user_create=user_in) user = User.create(session=session, create_obj=user_in)
if settings.emails_enabled and user_in.email: if settings.emails_enabled and user_in.email:
email_data = generate_new_account_email( email_data = generate_new_account_email(
email_to=user_in.email, username=user_in.email, password=user_in.password email_to=user_in.email, username=user_in.email, password=user_in.password
@@ -84,7 +82,7 @@ def update_user_me(
""" """
if user_in.email: if user_in.email:
existing_user = crud.get_user_by_email(session=session, email=user_in.email) existing_user = User.get_by_email(session=session, email=user_in.email)
if existing_user and existing_user.id != current_user.id: if existing_user and existing_user.id != current_user.id:
raise HTTPException( raise HTTPException(
status_code=409, detail="User with this email already exists" status_code=409, detail="User with this email already exists"
@@ -144,14 +142,14 @@ def register_user(session: SessionDep, user_in: UserRegister) -> Any:
""" """
Create new user without the need to be logged in. Create new user without the need to be logged in.
""" """
user = crud.get_user_by_email(session=session, email=user_in.email) user = User.get_by_email(session=session, email=user_in.email)
if user: if user:
raise HTTPException( raise HTTPException(
status_code=400, status_code=400,
detail="The user with this email already exists in the system", detail="The user with this email already exists in the system",
) )
user_create = UserCreate.model_validate(user_in) user_create = UserCreate.model_validate(user_in)
user = crud.create_user(session=session, user_create=user_create) user = User.create(session=session, create_obj=user_create)
return user return user
@@ -195,13 +193,13 @@ def update_user(
detail="The user with this id does not exist in the system", detail="The user with this id does not exist in the system",
) )
if user_in.email: if user_in.email:
existing_user = crud.get_user_by_email(session=session, email=user_in.email) existing_user = User.get_by_email(session=session, email=user_in.email)
if existing_user and existing_user.id != user_id: if existing_user and existing_user.id != user_id:
raise HTTPException( raise HTTPException(
status_code=409, detail="User with this email already exists" status_code=409, detail="User with this email already exists"
) )
db_user = crud.update_user(session=session, db_user=db_user, user_in=user_in) db_user = User.update(session=session, db_obj=db_user, in_obj=user_in)
return db_user return db_user
@@ -219,8 +217,8 @@ def delete_user(
raise HTTPException( raise HTTPException(
status_code=403, detail="Super users are not allowed to delete themselves" status_code=403, detail="Super users are not allowed to delete themselves"
) )
statement = delete(Item).where(col(Item.owner_id) == user_id) # statement = delete(Item).where(col(Item.owner_id) == user_id)
session.exec(statement) # type: ignore # session.exec(statement) # type: ignore
session.delete(user) session.delete(user)
session.commit() session.commit()
return Message(message="User deleted successfully") return Message(message="User deleted successfully")

View File

@@ -1,6 +1,7 @@
from sqlmodel import Session, create_engine, select from sqlmodel import Session, create_engine, select
from app.core.config import settings from app.core.config import settings
from app.models.user import User, UserCreate from app.models.user import User, UserCreate
engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI)) engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))
@@ -15,10 +16,10 @@ def init_db(session: Session) -> None:
# Tables should be created with Alembic migrations # Tables should be created with Alembic migrations
# But if you don't want to use migrations, create # But if you don't want to use migrations, create
# the tables un-commenting the next lines # the tables un-commenting the next lines
# from sqlmodel import SQLModel from app.models.base import BaseSQLModel
# This works because the models are already imported and registered from app.models # This works because the models are already imported and registered from app.models
# SQLModel.metadata.create_all(engine) BaseSQLModel.metadata.create_all(engine)
user = session.exec( user = session.exec(
select(User).where(User.email == settings.FIRST_SUPERUSER) select(User).where(User.email == settings.FIRST_SUPERUSER)
@@ -27,5 +28,7 @@ def init_db(session: Session) -> None:
user_in = UserCreate( user_in = UserCreate(
email=settings.FIRST_SUPERUSER, email=settings.FIRST_SUPERUSER,
password=settings.FIRST_SUPERUSER_PASSWORD, password=settings.FIRST_SUPERUSER_PASSWORD,
is_verified=True,
is_active=True,
) )
user = User.create_user(session=session, user_create=user_in) user = User.create(session=session, create_obj=user_in)

View File

@@ -19,11 +19,15 @@ class IsVerified(BaseModel):
class UserName(BaseModel): class UserName(BaseModel):
username: str | None = Field(default=None, nullable=True, unique=True, max_length=255) username: str | None = Field(
default=None, nullable=True, unique=True, max_length=255
)
class Email(BaseModel): class Email(BaseModel):
email: EmailStr | None = Field(default=None, nullable=True, unique=True, max_length=255) email: EmailStr | None = Field(
default=None, nullable=True, unique=True, max_length=255
)
class EmailUpdate(Email): class EmailUpdate(Email):
@@ -37,6 +41,7 @@ class ScoutingId(BaseModel):
class Password(BaseModel): class Password(BaseModel):
password: str = Field(min_length=8, max_length=100) password: str = Field(min_length=8, max_length=100)
class PasswordUpdate(Password): class PasswordUpdate(Password):
password: str | None = Field(default=None, min_length=8, max_length=40) password: str | None = Field(default=None, min_length=8, max_length=40)
@@ -48,5 +53,6 @@ class RowId(BaseModel):
default_factory=uuid.uuid4, default_factory=uuid.uuid4,
) )
class RowIdPublic(RowId): class RowIdPublic(RowId):
id: RowIdType id: RowIdType

View File

@@ -24,7 +24,7 @@ class UserBase(
mixin.ScoutingId, mixin.ScoutingId,
mixin.IsActive, mixin.IsActive,
mixin.IsVerified, mixin.IsVerified,
BaseSQLModel BaseSQLModel,
): ):
pass pass
@@ -119,3 +119,25 @@ class UsersPublic(BaseSQLModel):
# endregion # endregion
# region Password manager ######################################################
# JSON payload containing access token
class Token(BaseSQLModel):
access_token: str
token_type: str = "bearer"
# Contents of JWT token
class TokenPayload(BaseSQLModel):
sub: str | None = None
class NewPassword(BaseSQLModel):
token: str
new_password: str = Field(min_length=8, max_length=40)
# endregion

View File

@@ -5,8 +5,7 @@ from sqlmodel import Session
from app.core.config import settings from app.core.config import settings
from app.core.security import verify_password from app.core.security import verify_password
from app.crud import create_user from app.models.user import User, UserCreate
from app.models import UserCreate
from app.tests.utils.user import user_authentication_headers from app.tests.utils.user import user_authentication_headers
from app.tests.utils.utils import random_email, random_lower_string from app.tests.utils.utils import random_email, random_lower_string
from app.utils import generate_password_reset_token from app.utils import generate_password_reset_token
@@ -84,7 +83,7 @@ def test_reset_password(client: TestClient, db: Session) -> None:
is_active=True, is_active=True,
is_superuser=False, is_superuser=False,
) )
user = create_user(session=db, user_create=user_create) user = User.create(session=db, create_obj=user_create)
token = generate_password_reset_token(email=email) token = generate_password_reset_token(email=email)
headers = user_authentication_headers(client=client, email=email, password=password) headers = user_authentication_headers(client=client, email=email, password=password)
data = {"new_password": new_password, "token": token} data = {"new_password": new_password, "token": token}

View File

@@ -4,10 +4,9 @@ from unittest.mock import patch
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from sqlmodel import Session, select from sqlmodel import Session, select
from app import crud
from app.core.config import settings from app.core.config import settings
from app.core.security import verify_password from app.core.security import verify_password
from app.models import User, UserCreate from app.models.user import User, UserCreate
from app.tests.utils.utils import random_email, random_lower_string from app.tests.utils.utils import random_email, random_lower_string
@@ -51,7 +50,7 @@ def test_create_user_new_email(
) )
assert 200 <= r.status_code < 300 assert 200 <= r.status_code < 300
created_user = r.json() created_user = r.json()
user = crud.get_user_by_email(session=db, email=username) user = User.get_by_email(session=db, email=username)
assert user assert user
assert user.email == created_user["email"] assert user.email == created_user["email"]
@@ -62,7 +61,7 @@ def test_get_existing_user(
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
user_id = user.id user_id = user.id
r = client.get( r = client.get(
f"{settings.API_V1_STR}/users/{user_id}", f"{settings.API_V1_STR}/users/{user_id}",
@@ -70,7 +69,7 @@ def test_get_existing_user(
) )
assert 200 <= r.status_code < 300 assert 200 <= r.status_code < 300
api_user = r.json() api_user = r.json()
existing_user = crud.get_user_by_email(session=db, email=username) existing_user = User.get_by_email(session=db, email=username)
assert existing_user assert existing_user
assert existing_user.email == api_user["email"] assert existing_user.email == api_user["email"]
@@ -79,7 +78,7 @@ def test_get_existing_user_current_user(client: TestClient, db: Session) -> None
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
user_id = user.id user_id = user.id
login_data = { login_data = {
@@ -97,7 +96,7 @@ def test_get_existing_user_current_user(client: TestClient, db: Session) -> None
) )
assert 200 <= r.status_code < 300 assert 200 <= r.status_code < 300
api_user = r.json() api_user = r.json()
existing_user = crud.get_user_by_email(session=db, email=username) existing_user = User.get_by_email(session=db, email=username)
assert existing_user assert existing_user
assert existing_user.email == api_user["email"] assert existing_user.email == api_user["email"]
@@ -120,7 +119,7 @@ def test_create_user_existing_username(
# username = email # username = email
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
crud.create_user(session=db, user_create=user_in) User.create(session=db, create_obj=user_in)
data = {"email": username, "password": password} data = {"email": username, "password": password}
r = client.post( r = client.post(
f"{settings.API_V1_STR}/users/", f"{settings.API_V1_STR}/users/",
@@ -152,12 +151,12 @@ def test_retrieve_users(
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
crud.create_user(session=db, user_create=user_in) User.create(session=db, create_obj=user_in)
username2 = random_email() username2 = random_email()
password2 = random_lower_string() password2 = random_lower_string()
user_in2 = UserCreate(email=username2, password=password2) user_in2 = UserCreate(email=username2, password=password2)
crud.create_user(session=db, user_create=user_in2) User.create(session=db, create_obj=user_in2)
r = client.get(f"{settings.API_V1_STR}/users/", headers=superuser_token_headers) r = client.get(f"{settings.API_V1_STR}/users/", headers=superuser_token_headers)
all_users = r.json() all_users = r.json()
@@ -251,7 +250,7 @@ def test_update_user_me_email_exists(
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
data = {"email": user.email} data = {"email": user.email}
r = client.patch( r = client.patch(
@@ -326,7 +325,7 @@ def test_update_user(
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
data = {"full_name": "Updated_full_name"} data = {"full_name": "Updated_full_name"}
r = client.patch( r = client.patch(
@@ -365,12 +364,12 @@ def test_update_user_email_exists(
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
username2 = random_email() username2 = random_email()
password2 = random_lower_string() password2 = random_lower_string()
user_in2 = UserCreate(email=username2, password=password2) user_in2 = UserCreate(email=username2, password=password2)
user2 = crud.create_user(session=db, user_create=user_in2) user2 = User.create(session=db, create_obj=user_in2)
data = {"email": user2.email} data = {"email": user2.email}
r = client.patch( r = client.patch(
@@ -386,7 +385,7 @@ def test_delete_user_me(client: TestClient, db: Session) -> None:
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
user_id = user.id user_id = user.id
login_data = { login_data = {
@@ -431,7 +430,7 @@ def test_delete_user_super_user(
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
user_id = user.id user_id = user.id
r = client.delete( r = client.delete(
f"{settings.API_V1_STR}/users/{user_id}", f"{settings.API_V1_STR}/users/{user_id}",
@@ -458,7 +457,7 @@ def test_delete_user_not_found(
def test_delete_user_current_super_user_error( def test_delete_user_current_super_user_error(
client: TestClient, superuser_token_headers: dict[str, str], db: Session client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None: ) -> None:
super_user = crud.get_user_by_email(session=db, email=settings.FIRST_SUPERUSER) super_user = User.get_by_email(session=db, email=settings.FIRST_SUPERUSER)
assert super_user assert super_user
user_id = super_user.id user_id = super_user.id
@@ -476,7 +475,7 @@ def test_delete_user_without_privileges(
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
r = client.delete( r = client.delete(
f"{settings.API_V1_STR}/users/{user.id}", f"{settings.API_V1_STR}/users/{user.id}",

View File

@@ -1,9 +1,8 @@
from fastapi.encoders import jsonable_encoder from fastapi.encoders import jsonable_encoder
from sqlmodel import Session from sqlmodel import Session
from app import crud
from app.core.security import verify_password from app.core.security import verify_password
from app.models import User, UserCreate, UserUpdate from app.models.user import User, UserCreate, UserUpdate
from app.tests.utils.utils import random_email, random_lower_string from app.tests.utils.utils import random_email, random_lower_string
@@ -11,7 +10,7 @@ def test_create_user(db: Session) -> None:
email = random_email() email = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=email, password=password) user_in = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
assert user.email == email assert user.email == email
assert hasattr(user, "hashed_password") assert hasattr(user, "hashed_password")
@@ -20,8 +19,8 @@ def test_authenticate_user(db: Session) -> None:
email = random_email() email = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=email, password=password) user_in = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
authenticated_user = crud.authenticate(session=db, email=email, password=password) authenticated_user = User.authenticate(session=db, email=email, password=password)
assert authenticated_user assert authenticated_user
assert user.email == authenticated_user.email assert user.email == authenticated_user.email
@@ -29,7 +28,7 @@ def test_authenticate_user(db: Session) -> None:
def test_not_authenticate_user(db: Session) -> None: def test_not_authenticate_user(db: Session) -> None:
email = random_email() email = random_email()
password = random_lower_string() password = random_lower_string()
user = crud.authenticate(session=db, email=email, password=password) user = User.authenticate(session=db, email=email, password=password)
assert user is None assert user is None
@@ -37,7 +36,7 @@ def test_check_if_user_is_active(db: Session) -> None:
email = random_email() email = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=email, password=password) user_in = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
assert user.is_active is True assert user.is_active is True
@@ -45,7 +44,7 @@ def test_check_if_user_is_active_inactive(db: Session) -> None:
email = random_email() email = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=email, password=password, disabled=True) user_in = UserCreate(email=email, password=password, disabled=True)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
assert user.is_active assert user.is_active
@@ -53,7 +52,7 @@ def test_check_if_user_is_superuser(db: Session) -> None:
email = random_email() email = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=email, password=password, is_superuser=True) user_in = UserCreate(email=email, password=password, is_superuser=True)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
assert user.is_superuser is True assert user.is_superuser is True
@@ -61,7 +60,7 @@ def test_check_if_user_is_superuser_normal_user(db: Session) -> None:
username = random_email() username = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=username, password=password) user_in = UserCreate(email=username, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
assert user.is_superuser is False assert user.is_superuser is False
@@ -69,7 +68,7 @@ def test_get_user(db: Session) -> None:
password = random_lower_string() password = random_lower_string()
username = random_email() username = random_email()
user_in = UserCreate(email=username, password=password, is_superuser=True) user_in = UserCreate(email=username, password=password, is_superuser=True)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
user_2 = db.get(User, user.id) user_2 = db.get(User, user.id)
assert user_2 assert user_2
assert user.email == user_2.email assert user.email == user_2.email
@@ -80,11 +79,11 @@ def test_update_user(db: Session) -> None:
password = random_lower_string() password = random_lower_string()
email = random_email() email = random_email()
user_in = UserCreate(email=email, password=password, is_superuser=True) user_in = UserCreate(email=email, password=password, is_superuser=True)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
new_password = random_lower_string() new_password = random_lower_string()
user_in_update = UserUpdate(password=new_password, is_superuser=True) user_in_update = UserUpdate(password=new_password, is_superuser=True)
if user.id is not None: if user.id is not None:
crud.update_user(session=db, db_user=user, user_in=user_in_update) User.update(session=db, db_obj=user, in_obj=user_in_update)
user_2 = db.get(User, user.id) user_2 = db.get(User, user.id)
assert user_2 assert user_2
assert user.email == user_2.email assert user.email == user_2.email

View File

@@ -1,16 +0,0 @@
from sqlmodel import Session
from app import crud
from app.models import Item, ItemCreate
from app.tests.utils.user import create_random_user
from app.tests.utils.utils import random_lower_string
def create_random_item(db: Session) -> Item:
user = create_random_user(db)
owner_id = user.id
assert owner_id is not None
title = random_lower_string()
description = random_lower_string()
item_in = ItemCreate(title=title, description=description)
return crud.create_item(session=db, item_in=item_in, owner_id=owner_id)

View File

@@ -1,9 +1,8 @@
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from sqlmodel import Session from sqlmodel import Session
from app import crud
from app.core.config import settings from app.core.config import settings
from app.models import User, UserCreate, UserUpdate from app.models.user import User, UserCreate, UserUpdate
from app.tests.utils.utils import random_email, random_lower_string from app.tests.utils.utils import random_email, random_lower_string
@@ -23,7 +22,7 @@ def create_random_user(db: Session) -> User:
email = random_email() email = random_email()
password = random_lower_string() password = random_lower_string()
user_in = UserCreate(email=email, password=password) user_in = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in) user = User.create(session=db, create_obj=user_in)
return user return user
@@ -36,14 +35,14 @@ def authentication_token_from_email(
If the user doesn't exist it is created first. If the user doesn't exist it is created first.
""" """
password = random_lower_string() password = random_lower_string()
user = crud.get_user_by_email(session=db, email=email) user = User.get_by_email(session=db, email=email)
if not user: if not user:
user_in_create = UserCreate(email=email, password=password) user_in_create = UserCreate(email=email, password=password)
user = crud.create_user(session=db, user_create=user_in_create) user = User.create(session=db, create_obj=user_in_create)
else: else:
user_in_update = UserUpdate(password=password) user_in_update = UserUpdate(password=password)
if not user.id: if not user.id:
raise Exception("User id not set") raise Exception("User id not set")
user = crud.update_user(session=db, db_user=user, user_in=user_in_update) user = User.update(session=db, db_obj=user, in_obj=user_in_update)
return user_authentication_headers(client=client, email=email, password=password) return user_authentication_headers(client=client, email=email, password=password)

View File

@@ -1,49 +1,49 @@
services: services:
# Local services are available on their ports, but also available on: # # Local services are available on their ports, but also available on:
# http://api.localhost.tiangolo.com: backend # # http://api.localhost.tiangolo.com: backend
# http://dashboard.localhost.tiangolo.com: frontend # # http://dashboard.localhost.tiangolo.com: frontend
# etc. To enable it, update .env, set: # # etc. To enable it, update .env, set:
# DOMAIN=localhost.tiangolo.com # # DOMAIN=localhost.tiangolo.com
proxy: # proxy:
image: traefik:3.0 # image: traefik:3.0
volumes: # volumes:
- /var/run/docker.sock:/var/run/docker.sock # - /var/run/docker.sock:/var/run/docker.sock
ports: # ports:
- "80:80" # - "80:80"
- "8090:8080" # - "8090:8080"
# Duplicate the command from docker-compose.yml to add --api.insecure=true # # Duplicate the command from docker-compose.yml to add --api.insecure=true
command: # command:
# Enable Docker in Traefik, so that it reads labels from Docker services # # Enable Docker in Traefik, so that it reads labels from Docker services
- --providers.docker # - --providers.docker
# Add a constraint to only use services with the label for this stack # # Add a constraint to only use services with the label for this stack
- --providers.docker.constraints=Label(`traefik.constraint-label`, `traefik-public`) # - --providers.docker.constraints=Label(`traefik.constraint-label`, `traefik-public`)
# Do not expose all Docker services, only the ones explicitly exposed # # Do not expose all Docker services, only the ones explicitly exposed
- --providers.docker.exposedbydefault=false # - --providers.docker.exposedbydefault=false
# Create an entrypoint "http" listening on port 80 # # Create an entrypoint "http" listening on port 80
- --entrypoints.http.address=:80 # - --entrypoints.http.address=:80
# Create an entrypoint "https" listening on port 443 # # Create an entrypoint "https" listening on port 443
- --entrypoints.https.address=:443 # - --entrypoints.https.address=:443
# Enable the access log, with HTTP requests # # Enable the access log, with HTTP requests
- --accesslog # - --accesslog
# Enable the Traefik log, for configurations and errors # # Enable the Traefik log, for configurations and errors
- --log # - --log
# Enable debug logging for local development # # Enable debug logging for local development
- --log.level=DEBUG # - --log.level=DEBUG
# Enable the Dashboard and API # # Enable the Dashboard and API
- --api # - --api
# Enable the Dashboard and API in insecure mode for local development # # Enable the Dashboard and API in insecure mode for local development
- --api.insecure=true # - --api.insecure=true
labels: # labels:
# Enable Traefik for this service, to make it available in the public network # # Enable Traefik for this service, to make it available in the public network
- traefik.enable=true # - traefik.enable=true
- traefik.constraint-label=traefik-public # - traefik.constraint-label=traefik-public
# Dummy https-redirect middleware that doesn't really redirect, only to # # Dummy https-redirect middleware that doesn't really redirect, only to
# allow running it locally # # allow running it locally
- traefik.http.middlewares.https-redirect.contenttype.autodetect=false # - traefik.http.middlewares.https-redirect.contenttype.autodetect=false
networks: # networks:
- traefik-public # - traefik-public
- default # - default
db: db:
restart: "no" restart: "no"

View File

@@ -168,4 +168,4 @@ volumes:
networks: networks:
traefik-public: traefik-public:
# Allow setting it to false for testing # Allow setting it to false for testing
external: true # external: true