1
from datetime import timedelta
2

3
from pydantic_settings import BaseSettings
4

5
class AuthConfig(BaseSettings):
6
    JWT_ALG: str
7
    JWT_SECRET: str
8
    JWT_EXP: int = 5  # 分钟
9

10
    REFRESH_TOKEN_KEY: str
11
    REFRESH_TOKEN_EXP: timedelta = timedelta(days=30)
12

13
    SECURE_COOKIES: bool = True
14

15
auth_settings = AuthConfig()
16

17
# src.config
18
from pydantic import PostgresDsn, RedisDsn
19
from pydantic_settings import BaseSettings
20

21
from src.constants import Environment
22

23
class Config(BaseSettings):
24
    DATABASE_URL: PostgresDsn
25
    REDIS_URL: RedisDsn
26

27
    SITE_DOMAIN: str = "myapp.com"
28

29
    ENVIRONMENT: Environment = Environment.PRODUCTION
30

31
    SENTRY_DSN: str | None = None
32

33
    CORS_ORIGINS: list[str]
34
    CORS_ORIGINS_REGEX: str | None = None
35
    CORS_HEADERS: list[str]
36

37
    APP_VERSION: str = "1.0"
38

39
settings = Config()

1
async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
2
    post = await service.get_by_id(post_id)
3
    if not post:
4
        raise PostNotFound()
5

6
    return post
7

8
# router.py
9
@router.get("/posts/{post_id}", response_model=PostResponse)
10
async def get_post_by_id(post: dict[str, Any] = Depends(valid_post_id)):
11
    return post
12

13
@router.put("/posts/{post_id}", response_model=PostResponse)
14
async def update_post(
15
    update_data: PostUpdate,
16
    post: dict[str, Any] = Depends(valid_post_id),
17
):
18
    updated_post = await service.update(id=post["id"], data=update_data)
19
    return updated_post
20

21
@router.get("/posts/{post_id}/reviews", response_model=list[ReviewsResponse])
22
async def get_post_reviews(post: dict[str, Any] = Depends(valid_post_id)):
23
    post_reviews = await reviews_service.get_by_post_id(post["id"])
24
    return post_reviews

1
from fastapi.security import OAuth2PasswordBearer
2
import jwt  # PyJWT
3
from jwt.exceptions import InvalidTokenError
4

5
async def valid_post_id(post_id: UUID4) -> dict[str, Any]:
6
    post = await service.get_by_id(post_id)
7
    if not post:
8
        raise PostNotFound()
9

10
    return post
11

12
async def parse_jwt_data(
13
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
14
) -> dict[str, Any]:
15
    try:
16
        payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
17
    except InvalidTokenError:
18
        raise InvalidCredentials()
19

20
    return {"user_id": payload["id"]}
21

22
async def valid_owned_post(
23
    post: dict[str, Any] = Depends(valid_post_id),
24
    token_data: dict[str, Any] = Depends(parse_jwt_data),
25
) -> dict[str, Any]:
26
    if post["creator_id"] != token_data["user_id"]:
27
        raise UserNotOwner()
28

29
    return post
30

31
# router.py
32
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
33
async def get_user_post(post: dict[str, Any] = Depends(valid_owned_post)):
34
    return

1
from fastapi import BackgroundTasks
2
from fastapi.security import OAuth2PasswordBearer
3
import jwt  # PyJWT
4
from jwt.exceptions import InvalidTokenError
5

6
async def valid_post_id(post_id: UUID4) -> Mapping:
7
    post = await service.get_by_id(post_id)
8
    if not post:
9
        raise PostNotFound()
10

11
    return post
12

13
async def parse_jwt_data(
14
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
15
) -> dict:
16
    try:
17
        payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
18
    except InvalidTokenError:
19
        raise InvalidCredentials()
20

21
    return {"user_id": payload["id"]}
22

23
async def valid_owned_post(
24
    post: Mapping = Depends(valid_post_id),
25
    token_data: dict = Depends(parse_jwt_data),
26
) -> Mapping:
27
    if post["creator_id"] != token_data["user_id"]:
28
        raise UserNotOwner()
29

30
    return post
31

32
async def valid_active_creator(
33
    token_data: dict = Depends(parse_jwt_data),
34
):
35
    user = await users_service.get_by_id(token_data["user_id"])
36
    if not user["is_active"]:
37
        raise UserIsBanned()
38

39
    if not user["is_creator"]:
40
       raise UserNotCreator()
41

42
    return user
43

44

45
# router.py
46
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
47
async def get_user_post(
48
    worker: BackgroundTasks,
49
    post: Mapping = Depends(valid_owned_post),
50
    user: Mapping = Depends(valid_active_creator),
51
):
52
    """Get post that belong the active user."""
53
    worker.add_task(notifications_service.send_email, user["id"])
54
    return post

1
async def valid_profile_id(profile_id: UUID4) -> Mapping:
2
    profile = await service.get_by_id(profile_id)
3
    if not profile:
4
        raise ProfileNotFound()
5

6
    return profile
7

8
# src.creators.dependencies
9
async def valid_creator_id(profile: Mapping = Depends(valid_profile_id)) -> Mapping:
10
    if not profile["is_creator"]:
11
       raise ProfileNotCreator()
12

13
    return profile
14

15
# src.profiles.router.py
16
@router.get("/profiles/{profile_id}", response_model=ProfileResponse)
17
async def get_user_profile_by_id(profile: Mapping = Depends(valid_profile_id)):
18
    """Get profile by id."""
19
    return profile
20

21
# src.creators.router.py
22
@router.get("/creators/{profile_id}", response_model=ProfileResponse)
23
async def get_user_profile_by_id(
24
     creator_profile: Mapping = Depends(valid_creator_id)
25
):
26
    """Get creator's profile by id."""
27
    return creator_profile

1
from pydantic import BaseModel, field_validator
2

3
class ProfileCreate(BaseModel):
4
    username: str
5

6
    @field_validator("password", mode="after")
7
    @classmethod
8
    def valid_password(cls, password: str) -> str:
9
        if not re.match(STRONG_PASSWORD_PATTERN, password):
10
            raise ValueError(
11
                "Password must contain at least "
12
                "one lower character, "
13
                "one upper character, "
14
                "digit or "
15
                "special symbol"
16
            )
17

18
        return password
19

20
# src.profiles.routes
21
from fastapi import APIRouter
22

23
router = APIRouter()
24

25
@router.post("/profiles")
26
async def get_creator_posts(profile_data: ProfileCreate):
27
   pass

1
from typing import Any
2

3
from pydantic import UUID4
4
from sqlalchemy import desc, func, select, text
5
from sqlalchemy.sql.functions import coalesce
6

7
from src.database import database, posts, profiles, post_review, products
8

9
async def get_posts(
10
    creator_id: UUID4, *, limit: int = 10, offset: int = 0
11
) -> list[dict[str, Any]]:
12
    select_query = (
13
        select(
14
            (
15
                posts.c.id,
16
                posts.c.slug,
17
                posts.c.title,
18
                func.json_build_object(
19
                   text("'id', profiles.id"),
20
                   text("'first_name', profiles.first_name"),
21
                   text("'last_name', profiles.last_name"),
22
                   text("'username', profiles.username"),
23
                ).label("creator"),
24
            )
25
        )
26
        .select_from(posts.join(profiles, posts.c.owner_id == profiles.c.id))
27
        .where(posts.c.owner_id == creator_id)
28
        .limit(limit)
29
        .offset(offset)
30
        .group_by(
31
            posts.c.id,
32
            posts.c.type,
33
            posts.c.slug,
34
            posts.c.title,
35
            profiles.c.id,
36
            profiles.c.first_name,
37
            profiles.c.last_name,
38
            profiles.c.username,
39
            profiles.c.avatar,
40
        )
41
        .order_by(
42
            desc(coalesce(posts.c.updated_at, posts.c.published_at, posts.c.created_at))
43
        )
44
    )
45

46
    return await database.fetch_all(select_query)
47

48
# src.posts.schemas
49
from typing import Any
50

51
from pydantic import BaseModel, UUID4
52

53

54
class Creator(BaseModel):
55
    id: UUID4
56
    first_name: str
57
    last_name: str
58
    username: str
59

60
class Post(BaseModel):
61
    id: UUID4
62
    slug: str
63
    title: str
64
    creator: Creator
65

66

67
# src.posts.router
68
from fastapi import APIRouter, Depends
69

70
router = APIRouter()
71

72
@router.get("/creators/{creator_id}/posts", response_model=list[Post])
73
async def get_creator_posts(creator: dict[str, Any] = Depends(valid_creator_id)):
74
   posts = await service.get_posts(creator["id"])
75

76
   return posts