赞
踩
具体代码如下(main.py文件):
- #!/usr/bin/env python
- import os
- import sys
- from contextlib import asynccontextmanager
- from datetime import datetime
- from typing import List
-
- # python -m pip install --upgrade pip fastapi fastapi-cdn-host 'git+ssh://git@github.com/tortoise/tortoise-orm'
- import fastapi_cdn_host
- from fastapi import FastAPI, HTTPException, Request
- from fastapi.responses import RedirectResponse
- from pydantic import BaseModel
- from tortoise import fields, models
- from tortoise.contrib.fastapi import HTTPNotFoundError, RegisterTortoise
- from tortoise.contrib.pydantic import pydantic_model_creator
-
- __version__ = "1.0.0"
-
-
- class Users(models.Model):
- """
- The User model
- """
-
- id = fields.IntField(pk=True)
- #: This is a username
- username = fields.CharField(max_length=20, unique=True)
- name = fields.CharField(max_length=50, null=True)
- family_name = fields.CharField(max_length=50, null=True)
- category = fields.CharField(max_length=30, default="misc")
- password_hash = fields.CharField(max_length=128, null=True)
- created_at = fields.DatetimeField(auto_now_add=True)
- modified_at = fields.DatetimeField(auto_now=True)
-
- def full_name(self) -> str:
- """
- Returns the best name
- """
- if self.name or self.family_name:
- return f"{self.name or ''} {self.family_name or ''}".strip()
- return self.username
-
- class PydanticMeta:
- computed = ["full_name"]
- exclude = ["password_hash"]
-
-
- User_Pydantic = pydantic_model_creator(Users, name="User")
- UserIn_Pydantic = pydantic_model_creator(Users, name="UserIn", exclude_readonly=True)
-
-
- @asynccontextmanager
- async def lifespan(app: FastAPI):
- # 注册数据库、挂载Redis等……
- async with RegisterTortoise(
- app,
- db_url="sqlite://:memory:",
- modules={"models": ["__main__"]},
- generate_schemas=True,
- add_exception_handlers=True,
- ):
- yield
-
-
- app = FastAPI(title="FastAPI CDN Host Demo", lifespan=lifespan, version=__version__)
- fastapi_cdn_host.patch_docs(app)
-
-
- @app.get("/", include_in_schema=False)
- async def root(request: Request) -> RedirectResponse:
- """首页直接跳转到文档"""
- return RedirectResponse("/docs")
-
-
- @app.get("/robots.txt", include_in_schema=False)
- async def robots_txt():
- """声明不允许爬虫访问"""
- return """
- User-agent: *
- Disallow: /
- """
-
-
- @app.get("/app")
- async def app_info(request: Request) -> dict[str, str | dict | datetime | None]:
- """展示客户端IP、服务器时间等信息"""
- headers = dict(request.headers)
- ip = getattr(request.client, "host", "")
- url = request.url
- return {
- "your ip": ip,
- "version": __version__,
- "now": datetime.now(),
- "headers": headers,
- "path": request.url.path,
- "url": {"scheme": url.scheme, "hostname": url.hostname, "port": url.port},
- }
-
-
- class Status(BaseModel):
- message: str
-
-
- @app.get("/users", response_model=List[User_Pydantic])
- async def get_users():
- return await User_Pydantic.from_queryset(Users.all())
-
-
- @app.post("/users", response_model=User_Pydantic)
- async def create_user(user: UserIn_Pydantic):
- user_obj = await Users.create(**user.dict(exclude_unset=True))
- return await User_Pydantic.from_tortoise_orm(user_obj)
-
-
- @app.get(
- "/user/{user_id}",
- response_model=User_Pydantic,
- responses={404: {"model": HTTPNotFoundError}},
- )
- async def get_user(user_id: int):
- return await User_Pydantic.from_queryset_single(Users.get(id=user_id))
-
-
- @app.put(
- "/user/{user_id}",
- response_model=User_Pydantic,
- responses={404: {"model": HTTPNotFoundError}},
- )
- async def update_user(user_id: int, user: UserIn_Pydantic):
- await Users.filter(id=user_id).update(**user.model_dump(exclude_unset=True))
- return await User_Pydantic.from_queryset_single(Users.get(id=user_id))
-
-
- @app.delete(
- "/user/{user_id}",
- response_model=Status,
- responses={404: {"model": HTTPNotFoundError}},
- )
- async def delete_user(user_id: int):
- deleted_count = await Users.filter(id=user_id).delete()
- if not deleted_count:
- raise HTTPException(status_code=404, detail=f"User {user_id} not found")
- return Status(message=f"Deleted user {user_id}")
-
-
- def _runserver() -> int:
- """This is for debug mode to start server. For prod, use supervisor+gunicorn instead."""
- return os.system("fastapi dev")
-
-
- if __name__ == "__main__": # pragma: no cover
- sys.exit(_runserver())
创建虚拟环境、安装依赖、执行格式化检查
- python3.12 -m venv venv
- source venv/*/activate
- python -m pip install --upgrade pip fastapi fastapi-cdn-host 'git+ssh://git@github.com/tortoise/tortoise-orm' 'fast-dev-cli[all]'
- fast lint main.py
报错如下:
- 1 file left unchanged
- main.py:105: error: Variable "main.User_Pydantic" is not valid as a type [valid-type]
- main.py:105: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases
- main.py:111: error: Variable "main.UserIn_Pydantic" is not valid as a type [valid-type]
- main.py:111: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases
- main.py:112: error: UserIn_Pydantic? has no attribute "dict" [attr-defined]
- main.py:130: error: Variable "main.UserIn_Pydantic" is not valid as a type [valid-type]
- main.py:130: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases
- main.py:131: error: UserIn_Pydantic? has no attribute "model_dump" [attr-defined]
- Found 5 errors in 1 file (checked 1 source file)
解决方法是在type checking的时候,用class的方式定义xxx_Pydantic:
- from typing import TYPE_CHECKING
-
-
- if TYPE_CHECKING:
- from tortoise.contrib.pydantic import PydanticModel
-
- class UserIn_Pydantic(Users, PydanticModel): # type:ignore[misc]
- pass
-
- class User_Pydantic(Users, PydanticModel): # type:ignore[misc]
- pass
- else:
- User_Pydantic = pydantic_model_creator(Users, name="User")
- UserIn_Pydantic = pydantic_model_creator(
- Users, name="UserIn", exclude_readonly=True
- )
修改后的完整文件如下:
- #!/usr/bin/env python
- import os
- import sys
- from contextlib import asynccontextmanager
- from datetime import datetime
- from typing import TYPE_CHECKING, List
-
- # python -m pip install --upgrade pip fastapi fastapi-cdn-host 'git+ssh://git@github.com/tortoise/tortoise-orm'
- import fastapi_cdn_host
- from fastapi import FastAPI, HTTPException, Request
- from fastapi.responses import RedirectResponse
- from pydantic import BaseModel
- from tortoise import fields, models
- from tortoise.contrib.fastapi import HTTPNotFoundError, RegisterTortoise
- from tortoise.contrib.pydantic import pydantic_model_creator
-
- __version__ = "1.0.0"
-
-
- class Users(models.Model):
- """
- The User model
- """
-
- id = fields.IntField(pk=True)
- #: This is a username
- username = fields.CharField(max_length=20, unique=True)
- name = fields.CharField(max_length=50, null=True)
- family_name = fields.CharField(max_length=50, null=True)
- category = fields.CharField(max_length=30, default="misc")
- password_hash = fields.CharField(max_length=128, null=True)
- created_at = fields.DatetimeField(auto_now_add=True)
- modified_at = fields.DatetimeField(auto_now=True)
-
- def full_name(self) -> str:
- """
- Returns the best name
- """
- if self.name or self.family_name:
- return f"{self.name or ''} {self.family_name or ''}".strip()
- return self.username
-
- class PydanticMeta:
- computed = ["full_name"]
- exclude = ["password_hash"]
-
-
- if TYPE_CHECKING:
- from tortoise.contrib.pydantic import PydanticModel
-
- class UserIn_Pydantic(Users, PydanticModel): # type:ignore[misc]
- pass
-
- class User_Pydantic(Users, PydanticModel): # type:ignore[misc]
- pass
- else:
- User_Pydantic = pydantic_model_creator(Users, name="User")
- UserIn_Pydantic = pydantic_model_creator(
- Users, name="UserIn", exclude_readonly=True
- )
-
-
- @asynccontextmanager
- async def lifespan(app: FastAPI):
- # 注册数据库、挂载Redis等……
- async with RegisterTortoise(
- app,
- db_url="sqlite://:memory:",
- modules={"models": ["__main__"]},
- generate_schemas=True,
- add_exception_handlers=True,
- ):
- yield
-
-
- app = FastAPI(title="FastAPI CDN Host Demo", lifespan=lifespan, version=__version__)
- fastapi_cdn_host.patch_docs(app)
-
-
- @app.get("/", include_in_schema=False)
- async def root(request: Request) -> RedirectResponse:
- """首页直接跳转到文档"""
- return RedirectResponse("/docs")
-
-
- @app.get("/robots.txt", include_in_schema=False)
- async def robots_txt():
- """声明不允许爬虫访问"""
- return """
- User-agent: *
- Disallow: /
- """
-
-
- @app.get("/app")
- async def app_info(request: Request) -> dict[str, str | dict | datetime | None]:
- """展示客户端IP、服务器时间等信息"""
- headers = dict(request.headers)
- ip = getattr(request.client, "host", "")
- url = request.url
- return {
- "your ip": ip,
- "version": __version__,
- "now": datetime.now(),
- "headers": headers,
- "path": request.url.path,
- "url": {"scheme": url.scheme, "hostname": url.hostname, "port": url.port},
- }
-
-
- class Status(BaseModel):
- message: str
-
-
- @app.get("/users", response_model=List[User_Pydantic])
- async def get_users():
- return await User_Pydantic.from_queryset(Users.all())
-
-
- @app.post("/users", response_model=User_Pydantic)
- async def create_user(user: UserIn_Pydantic):
- user_obj = await Users.create(**user.dict(exclude_unset=True))
- return await User_Pydantic.from_tortoise_orm(user_obj)
-
-
- @app.get(
- "/user/{user_id}",
- response_model=User_Pydantic,
- responses={404: {"model": HTTPNotFoundError}},
- )
- async def get_user(user_id: int):
- return await User_Pydantic.from_queryset_single(Users.get(id=user_id))
-
-
- @app.put(
- "/user/{user_id}",
- response_model=User_Pydantic,
- responses={404: {"model": HTTPNotFoundError}},
- )
- async def update_user(user_id: int, user: UserIn_Pydantic):
- await Users.filter(id=user_id).update(**user.model_dump(exclude_unset=True))
- return await User_Pydantic.from_queryset_single(Users.get(id=user_id))
-
-
- @app.delete(
- "/user/{user_id}",
- response_model=Status,
- responses={404: {"model": HTTPNotFoundError}},
- )
- async def delete_user(user_id: int):
- deleted_count = await Users.filter(id=user_id).delete()
- if not deleted_count:
- raise HTTPException(status_code=404, detail=f"User {user_id} not found")
- return Status(message=f"Deleted user {user_id}")
-
-
- def _runserver() -> int:
- """This is for debug mode to start server. For prod, use supervisor+gunicorn instead."""
- return os.system("fastapi dev")
-
-
- if __name__ == "__main__": # pragma: no cover
- sys.exit(_runserver())
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。