Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # app.py
- from http import HTTPStatus
- from fastapi import FastAPI, HTTPException
- from sqlalchemy import create_engine, select
- from sqlalchemy.orm import Session
- from fast_zero.models import User
- from fast_zero.schemas import Message, UserDB, UserList, UserPublic, UserSchema
- from fast_zero.settings import Settings
- app = FastAPI()
- database = [] # Lista provisória para fins de estudo
- @app.get('/', status_code=HTTPStatus.OK, response_model=Message)
- def read_root():
- return {'message': 'Olá Mundo!'}
- @app.post('/users/', response_model=UserPublic, status_code=HTTPStatus.CREATED)
- def create_user(user: UserSchema):
- # user_with_id = UserDB(**user.model_dump(), id=len(database) + 1)
- # database.append(user_with_id)
- # return user_with_id
- engine = create_engine(Settings().DATABASE_URL)
- with Session(engine) as session:
- db_user = session.scalar(
- select(User).where(
- (User.username == user.username) | (User.email == user.email)
- )
- )
- if db_user:
- if db_user.username == user.username:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail='Username already exists!',
- )
- elif db_user.email == user.email:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail='Email already exists!',
- )
- db_user = User(
- username=user.username, email=user.email, password=user.password
- )
- session.add(db_user)
- session.commit()
- session.refresh(db_user)
- return db_user
- @app.get('/users/', response_model=UserList)
- def read_users():
- return {'users': database}
- @app.get('/users/{user_id}', response_model=UserPublic)
- def read_user_exercicio(user_id: int):
- if user_id > len(database) or user_id < 1:
- raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND, detail='User not found!'
- )
- return database[user_id - 1]
- @app.put('/users/{user_id}', response_model=UserPublic)
- def update_user(user_id: int, user: UserSchema):
- if user_id < 1 or user_id > len(database):
- raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND, detail='User not found!'
- )
- user_with_id = UserDB(id=user_id, **user.model_dump())
- database[user_id - 1] = user_with_id
- return user_with_id
- @app.delete('/users/{user_id}', response_model=Message)
- def delete_user(user_id: int):
- if user_id < 1 or user_id > len(database):
- raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND, detail='User not found!'
- )
- del database[user_id - 1]
- return {'message': 'User deleted!'}
- #####################################################################################
- # errors
- INFO 127.0.0.1:55195 - "POST /users/ HTTP/1.1" 500
- ERROR Exception in ASGI application
- Traceback (most recent call last):
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\uvicorn\protocols\http\httptools_impl.py", line 409, in run_asgi
- result = await app( # type: ignore[func-returns-value]
- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- self.scope, self.receive, self.send
- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- )
- ^
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\uvicorn\middleware\proxy_headers.py", line 60, in __call__
- return await self.app(scope, receive, send)
- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\fastapi\applications.py", line 1054, in __call__
- await super().__call__(scope, receive, send)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\applications.py", line 112, in __call__
- await self.middleware_stack(scope, receive, send)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\middleware\errors.py", line 187, in __call__
- raise exc
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\middleware\errors.py", line 165, in __call__
- await self.app(scope, receive, _send)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\middleware\exceptions.py", line 62, in __call__
- await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
- raise exc
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\_exception_handler.py", line 42, in wrapped_app
- await app(scope, receive, sender)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\routing.py", line 715, in __call__
- await self.middleware_stack(scope, receive, send)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\routing.py", line 735, in app
- await route.handle(scope, receive, send)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\routing.py", line 288, in handle
- await self.app(scope, receive, send)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\routing.py", line 76, in app
- await wrap_app_handling_exceptions(app, request)(scope, receive, send)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
- raise exc
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\_exception_handler.py", line 42, in wrapped_app
- await app(scope, receive, sender)
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\routing.py", line 73, in app
- response = await f(request)
- ^^^^^^^^^^^^^^^^
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\fastapi\routing.py", line 301, in app
- raw_response = await run_endpoint_function(
- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- ...<3 lines>...
- )
- ^
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\fastapi\routing.py", line 214, in run_endpoint_function
- return await run_in_threadpool(dependant.call, **values)
- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\starlette\concurrency.py", line 37, in run_in_threadpool
- return await anyio.to_thread.run_sync(func)
- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\anyio\to_thread.py", line 56, in run_sync
- return await get_async_backend().run_sync_in_worker_thread(
- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- func, args, abandon_on_cancel=abandon_on_cancel, limiter=limiter
- ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
- )
- ^
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\anyio\_backends\_asyncio.py", line 2461, in run_sync_in_worker_thread
- return await future
- ^^^^^^^^^^^^
- File "E:\python\projetos\fast_zero\.venv\Lib\site-packages\anyio\_backends\_asyncio.py", line 962, in run
- result = context.run(func, *args)
- File "E:\python\projetos\fast_zero\fast_zero\app.py", line 23, in create_user
- def create_user(user: UserSchema):
- ^^^^
- NameError: name 'crea' is not defined
Advertisement