fastapi-practices / fastapi_best_architecture Goto Github PK

Remove the existing response_200 method and reference and replace it with the success method

New routing system：From #106 (comment)

可以考虑用amis来前端

The conflict is caused by:
    The user requested starlette==0.27.0
    fastapi 0.95.0 depends on starlette<0.27.0 and >=0.26.1

To fix this you could try to:
1. loosen the range of package versions you've specified
2. remove package versions to allow pip attempt to solve the dependency conflict

https://github.com/tiangolo/fastapi/releases/tag/0.95.2

From #4 (reply in thread)

Add token refreshing mechanism (may be related to front-end work)

When creating a department, if the mailbox format and status are incorrect, an exception will be thrown:

{
  "code": 500,
  "msg": "'permitted'",
  "data": null
}

Hi, @wu-clan

Missing crucial time data, is there a way to customize the output?

INFO:     127.0.0.1:0 - "GET /api/v1/menus/sidebar HTTP/1.0" 200 OK

The addition of RBAC is the main one and will include some other additions and modifications

Please note that this will be not a forward compatible change

the default refresh time is the specified expiration time, add custom expiration time parameter, you can customize the expiration time

Form #4 (reply in thread)

@staticmethod
async def get_user_info(username: str):
    async with async_db_session() as db:
        user = await UserDao.get_user_by_username(db, username)
        if not user:
            raise errors.NotFoundError(msg='用户不存在')
    if user.avatar is not None:
        user.avatar = cut_path(AvatarPath + user.avatar)[1]
    # here
    return user

When deleting roles in batches, it is necessary to determine whether there is associated data

Need to judge whether there is data before deleting?

If there is no parent id, then the level is 1. If there is a parent, then you need to query the parent's level, and then add 1.

new_obj.update({'level': obj.parent_id + 1 if obj.parent_id else 1, 'create_user': user_id})

I understand that level is used to limit the number of layers created, but there are many complicated situations to consider here, such as updating the third level to another level, and the sub-levels under the third level should also change accordingly. etc.

When I was testing the task of sending an email, an exception occurred:

@celery_app.task
def send_active_email(email: str, active_url: str) -> str:
    send_verification_code_email(to=email, active_url=active_url)
    print(f'发送激活邮件到 {email} 成功')
    return 'Success'

KeyError: 'backend.app.tasks.send_active_email'

Request interface: /api/v1/mixes/tests/send

KeyError: 'backend.app.tasks.task_demo_async'

Received unregistered task of type 'backend.app.tasks.task_demo_async'.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you're using relative imports?

Please see
https://docs.celeryq.dev/en/latest/internals/protocol.html
for more information.

The traversal_to_tree algorithm has a bug, which has a lot to do with the order. If the parent is behind the child, the data will be empty.

in test_login function , only use username and password can pass the test?

https://docs.pydantic.dev/dev-v2/migration/

If you turn on AccessMiddleware settings, an exception "'async for' requires an object with aiter method, got method" will occur. Besides that, should request.state.ip be set in OperaLogMiddleware? Since OperaLogMiddleware is an optional plugin, if the user disables this plugin, the auth module will throw an exception for missing state.ip because it is used by captcha.

Hi, @downdawn

I've looked at the class view implementation of fastapi_utils in general, but I'm not very interested

# End Setup
app = FastAPI()
router = InferringRouter()  # Step 1: Create a router


@cbv(router)  # Step 2: Create and decorate a class to hold the endpoints
class ItemCBV:
    # Step 3: Add dependencies as class attributes
    session: Session = Depends(get_db)
    user_id: UserID = Depends(get_jwt_user)

    @router.post("/item")
    def create_item(self, item: ItemCreate) -> ItemInDB:
        # Step 4: Use `self.<dependency_name>` to access shared dependencies
        item_orm = ItemORM(name=item.name, owner=self.user_id)
        self.session.add(item_orm)
        self.session.commit()
        return ItemInDB.from_orm(item_orm)

For session, since we're using SQLAlchemy-2.0, we'll use async_db_session.begin() directly, this seems to make more sense

from functools import wraps

def session_decorator(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        async with async_db_session.begin() as db:
            return await func(db, *args, **kwargs)
    return wrapper


class ApiService:

    @staticmethod
    @session_decorator
    async def get_api_list(db):
        return await paginate(db, ApiDao.get_all_apis())

For Depends(get_jwt_user), we need to make FastAPI >= 0.95.0, then use the Annotated feature, and call it globally, E.g:

CurrentUser = Annotated[User, Depends(get_current_user)]

Like in the fastapi documentation: Share Annotated dependencies

When an exception occurs, the length of 255 is not enough, you can change it to TEXT

msg = getattr(e, 'msg', traceback.format_exc() if settings.ENVIRONMENT == 'dev' else 'Internal Server Error')

msg: Mapped[str | None] = mapped_column(String(255), comment='提示消息')

当新建用户时，遇到如下报错
File "asyncmy\converters.pyx", line 26, in asyncmy.converters.escape_item
val = encoder(val, mapping)
TypeError: Argument 'value' has incorrect type (expected str, got UUID)

asyncmy 0.2.8
Mysql 8.1.0

Discussed in #38

status: Annotated[StatusEnum | None, Query()] = None,

exception

{
  "code": 422,
  "msg": "请求参数非法",
  "data": {
    "errors": [
      {
        "loc": [
          "query",
          "method"
        ],
        "msg": "value is not a valid enumeration member; permitted: 0, 1, 2",
        "type": "type_error.enum",
        "ctx": {
          "enum_values": [
            0,
            1,
            2
          ]
        }
      }
    ]
  }
}

Discussed in #22

The get request sets the enumeration value and throws an exception

async def get_all_menus(
    title: Annotated[str | None, Query()] = None,
    status: Annotated[StatusType | None, Query()] = None,
):

{
  "code": 422,
  "msg": "请求参数非法",
  "data": {
    "errors": [
      {
        "loc": [
          "query",
          "status"
        ],
        "msg": "value is not a valid enumeration member; permitted: 0, 1",
        "type": "type_error.enum",
        "ctx": {
          "enum_values": [
            0,
            1
          ]
        }
      }
    ]
  }
}

The post request set enumeration value is invalid

For example, when creating a menu, although the MenuType is set, it is still possible to set data exceeding the limit

windows10

    total = cpu_times.user + cpu_times.nice + cpu_times.system + cpu_times.idle \
AttributeError: 'scputimes' object has no attribute 'nice'

1、Trying to log in, the background task creation login log is not executed, and the database has no data.

2、An exception occurs when logging in in json format

UnboundLocalError: local variable 'login_logs_params' referenced before assignment

https://github.com/zzzeek/sqlalchemy/blob/main/lib/sqlalchemy/exc.py

code type may be a string

Msg may not need to report error stack information?

Mysql5.7 uses the uuid type, and exceptions will occur during database migration.

default_factory=uuid4

From #43

from wu-clan#4 (comment)

Need to wait #15

The default form login return structure must be:

{"access_token": access_token, "token_type": "bearer"}

Otherwise, the swagger ui cannot request correctly, even if the verification passes

expectation is

{
  "code": 200,
  "msg": "Success",
  "data": {
        {"access_token": access_token, "token_type": "bearer"}
}
}

http://127.0.0.1:8000/v1/docs

Discussed in #44

Interim optimization solution with the ultimate goal of using custom interface permission field identifiers

It's better to provide a standardized return. The default TokenError is too general.

  raise HTTPException(
      status_code=HTTP_401_UNAUTHORIZED,
      detail="Not authenticated",
      headers={"WWW-Authenticate": "Bearer"},
  )

When the API request parameters are missing and request.json() fails to parse, proper error handling should be implemented.

            body_data = await request.body()
            if body_data:
                try:
                    json_data = await request.json()
                    args.update(json_data)
                except Exception as e:
                    log.exception(e)
                    args.update({'body': body_data.decode('utf-8')})

In order to use certain features, it was decided to use middleware to implement jwt authentication

Note: Make sure swagger authentication is available

From #4

from #19

Adding decorator methods

This is a common feature that we have overlooked in the current implementation.

感谢分享，我实践了做了个爬虫项目，写起来项目结构清晰和部署简单。

但是刚做完就更新3.10后版本了，想问下还会进行大的改版吗？尤其是在项目结构上。

Discussed in #2

This is an expected feature, but there is no good implementation plan.