안녕하세요. FastAPI로 개발을 하는 중, 로그 저장에 대해서 생각하는 시간을 좀 가져봤습니다.저는 요청, 응답, 쿼리 이렇게 3가지로 나누어서 일자별로 각각의 파일에 저장을 하려고 하는데요. 혹시 FastAPI에서 자체적으로 지원하는 기능중에 이렇게 로그를 나누어 저장하는 기능이 있을까요? 감사합니다.

학습 중에 fastapi는 get과 post만 지원하기 때문에 별도로 미들웨어를 이용해 put, patch, delete를 구현해야 한다고 이해를 하게 되었습니다.헌데 다른 학습 웹 문서나, gpt에게 여러 요청을 통해 확인을 하는 과정에서 app. 혹은 @router. 을 사용해 이미 지원되는 put, patch, delete를 사용할 수 있다고 설명을 하는 것을 확인할 수 있었습니다.제가 학습 영상에 대해 이해를 잘못하고 있는 것인지 혹 놓친게 있는지 알고 싶어 질문을 드립니다.

자동완성 되시던데요. 어떤 확장프로그램쓰시는걸까요

- 학습 관련 질문을 남겨주세요. 상세히 작성하면 더 좋아요! - 먼저 유사한 질문이 있었는지 검색해보세요. - 서로 예의를 지키며 존중하는 문화를 만들어가요. - 잠깐! 인프런 서비스 운영 관련 문의는 1:1 문의하기를 이용해주세요. from app.config import NAVER_API_ID, NAVER_API_SECRET가 아닌from app.config import get_secret로 import 한 이유가 궁금합니다. 클래스 내에 NAVER_API_ID = get_secret("NAVER_API_ID") NAVER_API_SECRET = get_secret("NAVER_API_SECRET")로 변수 선언한 이유가 있을까요?get_secret은 config 파일에서 이미 수행해 NAVER_API_ID와 NAVER_API_SECRET을 변수로 가지고 있으니,위처럼 클래스 내에 변수롤 또 선언하면 config 파일에서 get_secret 함수를 이용해 변수를 만든 이유가 없다고 생각이 들었습니다. from app.config import NAVER_API_ID, NAVER_API_SECRETconfig 파일에서 NAVER_API_ID와 NAVER_API_SECRET을 import해서 바로 사용하면 어떤 차이가 있나요?

(강의 색션 7_8 - 14:30)안녕하세요 선생님.좋은 강의 만들어 공유해 주셔서 진심으로 감사드립니다.Pydantic 모델 초기화 과정에서 아래의 코드처럼 Optional이 설정하지 않으면 Swagger UI에서 입력 값이 null이여서 엔티티 타입 불일치로 422 에러가 발생합니다.또한, parse_user_form에서 Optional 체크를 하고 Item에서는 별도로 설정하지 않으면 같은 에러가 발생하더군요.Q) FormData 부분은 Optional 체크를 하지 않아도 되고, pydantic model에서는 반드시 Optional 체크를 해야지만 문제 없이 정상 동작하는 이유에 대한 보충 설명 부탁드려도 될까요? 주석한 부분이 기존 코드입니다. 주석한 부분으로 코드를 변경하면 에러가 발생합니다.class Item(BaseModel): name: str = Field(..., min_length=2, max_length=50) # description: str = Field(None, max_length=500) description: Optional[str] = Field(None, max_length=500) price: float = Field(..., ge=0) # tax: float = None tax: Optional[float] = None @model_validator(mode='after') def tax_must_be_less_than_price(cls, values): price = values.price tax = values.tax # if tax > price: # tax가 NoneType이라 에러가 남. if tax is not None and tax > price: raise ValueError("Tax must be less then price") return values감사합니다.

현재 "신규 Blog 글 생성하기 - 01" 강의를 수강 중입니다.강의 중에, 쿼리의 commit() 실패시 rollback()이 자동으로 수행된다고 말씀 하셨는데,코드상에 with를 사용한 것도 아니고, context_get_conn()의 에러 발생시 rollback() 호출도 하지 않는데rollback()이 자동으로 수행되는게 맞는가요?close를 하면 자동으로 rollback() 된다고 하시는데close는 성공시에도 호출하는 것으로 코드를 이해하고 있습니다.insert, update의 성공시에도 rollback()이 되는게 맞는지도 알고 싶습니다.

Path(), Query(), Form() 모두 Depends에 사용이 가능할까요? 물론 BaseModel에는 인자와 동일한 이름의 클래스 변수가 정의되어 있다는 가정에서 의미합니다.form의 일부 인자들만으로 구성되게 사용이 가능할까요?함수의 입력 인자로 정의된 값이, Depends에서도 중복으로 사용이 가능한가요?

ValidationError가 valueerror보다 인스턴스 생성하기가 쉽지 않기 때문에 valueerror를 사용하셨다고 말씀하신 부분이 있는데, 어떤 부분의 어려움이 있는 것인지 좀 더 자세히 알고 싶어 질문을 드립니다. 검색으로 해당 부분에 대한 내용을 찾아 보려 했으나 잘 이해하지 못한 상태인 제가 선정할 수 있는 검색어들 자체가 모호하여 정확한 결과를 얻기가 힘들었습니다. 

이렇게 인터프리터랑 main.py 파일을 알맞게 사용하는데, Internal Server Error가 뜨고 컴퓨터를 재부팅하면 정상적으로 실행이 되는데 혹시 무슨 오류인지 알 수 있을까요?

안녕하세요 강사님강의 열심히 잘 듣고 있습니다.덕분에 FastAPI에 대한 재미를 더 많이 가지고 있어 감사할 따름입니다. lifespan 적용하신 강의 관련해서 질문이 있습니다.저는 현재 환경을 강의 내용과 다르게 진행하고있습니다. mysql이 아닌 postgresql로 진행하고 있고,postgresql + asyncpg를 이용하여 비동기 처리를 하고있습니다. lifespan 부분을 적용 시 아래 오류가 발생되고있습니다. Exception terminating connection <AdaptedConnection <asyncpg.connection.Connection object at 0x00000257982897B0>>Traceback (most recent call last): File "D:\personnel\Python\fastapi\fastapi_pguide-main\fastapi_pguide-main\AsyncDB_Handling\.venv\lib\site-packages\sqlalchemy\pool\base.py", line 374, in closeconnection self._dialect.do_terminate(connection) File "D:\personnel\Python\fastapi\fastapi_pguide-main\fastapi_pguide-main\AsyncDB_Handling\.venv\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 1117, in do_terminate dbapi_connection.terminate() File "D:\personnel\Python\fastapi\fastapi_pguide-main\fastapi_pguide-main\AsyncDB_Handling\.venv\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 910, in terminate self._connection.terminate() File "D:\personnel\Python\fastapi\fastapi_pguide-main\fastapi_pguide-main\AsyncDB_Handling\.venv\lib\site-packages\asyncpg\connection.py", line 1515, in terminate self._abort() File "D:\personnel\Python\fastapi\fastapi_pguide-main\fastapi_pguide-main\AsyncDB_Handling\.venv\lib\site-packages\asyncpg\connection.py", line 1567, in _abort self._protocol.abort() File "asyncpg\\protocol\\protocol.pyx", line 608, in asyncpg.protocol.protocol.BaseProtocol.abort File "C:\Users\user\AppData\Local\Programs\Python\Python310\lib\asyncio\selector_events.py", line 686, in abort self._force_close(None) File "C:\Users\user\AppData\Local\Programs\Python\Python310\lib\asyncio\selector_events.py", line 737, in forceclose self._loop.call_soon(self._call_connection_lost, exc) File "C:\Users\user\AppData\Local\Programs\Python\Python310\lib\asyncio\base_events.py", line 753, in call_soon self._check_closed() File "C:\Users\user\AppData\Local\Programs\Python\Python310\lib\asyncio\base_events.py", line 515, in checkclosed raise RuntimeError('Event loop is closed')RuntimeError: Event loop is closedThe garbage collector is trying to clean up non-checked-in connection <AdaptedConnection <asyncpg.connection.Connection object at 0x00000257982897B0>>, which will be terminated. Please ensure that SQLAlchemy pooled connections are returned to the pool explicitly, either by calling ``close()`` or by using appropriate context managers to manage their lifecycle.sys:1: SAWarning: The garbage collector is trying to clean up non-checked-in connection <AdaptedConnection <asyncpg.connection.Connection object at 0x00000257982897B0>>, which will be terminated. Please ensure that SQLAlchemy pooled connections are returned to the pool explicitly, either by calling ``close()`` or by using appropriate context managers to manage their lifecycle. 위와 같은 오류가 발생되고 있습니다. 방법을 찾아보려 gpt에게 문의도 해봤지만 강의 내용과 동일하거나 또는 starting, shutting (구 FastAPI)방식을 알려주고 있습니다.  나름 응용을 해보겠다고 진행한 건데.. 번거로운 질문 드려 죄송합니다.ㅜㅜ 제가 구성한 환경은 이와 같이 진행하였습니다. 참고부탁드리겠습니다. python =="^3.10" fastapi == "0.115.4" uvicorn == "0.32.0" python-multipart == "0.0.17" python-dotenv == "1.0.1" jinja2 == "3.1.4" sqlalchemy == "2.0.36" psycopg == {extras = ["binary", "pool"], version = "^3.2.3"} asyncpg == "^0.30.0" aiofiles == "24.1.0"감사합니다.

 PATCH API를 통해 todo를 수정하는 코드를 적용하고 실행시켰을 때, 응답의 Response body에 null이 찍힙니다..원인을 찾지 못해 질문드립니다.

안녕하세요,Redis Pub/Sub 수강 중위와 같이 메시지 브로커에 문제가 있는 것 같은데, 백엔드 지식이 부족해서 정확히 어떤 부분에서부터 원인을 찾아야할지 모르겠습니다.서버/컨테이너 재시작 해봤는데 동일한 걸로 봐서 다른 문제인 것 같아서 도움 요청드립니다.https://github.com/wozlsla/fastapi-async/blob/prac/src/shared/message_broker.py

선생님 강의에서는 아나콘다를 사용하셨는데, 저는 pipenv를 사용하고싶거든요..콘다 말고 다른 가상환경을 사용해도 괜찮을까요?

강의 정말 도움 많이 되었습니다!!강의 내용 중에 동시성인지 병렬성인지 헷갈리는게 있어 질문드립니다..!7:35에 일반 def 함수 라우터로 작성한 ThreadPool(병렬) 방식은 매번 요청시 ThreadPool에서 유휴 Tread를 찾아 동작하고 i/o wait가 발생할때 context switch를 해가면서 작업하는 것으로 이해했는데...이게 맞다면..! 완벽한 병렬성이 아니라 동시에 작업하는 것 처럼 보이는 동시성으로 동작하는 건가요?

안녕하세요,API와 DB 연결을 다루는 강의를 시청 중 강의 11분 50초 정도에 웹소켓 연결이 안 되는 문제가 생겨 질문 드립니다!코드를 전부 똑같이 따라 작성하고 있음에도 왜 localhost 7071로 경로 수정 후 연결이 되지 않는지 모르겠습니다..해결 방안이 있을지 여쭙고 싶습니다!  

안녕하세요, 강의 너무 잘 듣고 있습니다.잘 따라가고 있던 와중 프론트 코드에 URL 상수를 지정하는 과정들을 거치고 UI가 있는 창에서 개발자 도구를 열어도 연결 완료 문구가 뜨지 않아 글 남깁니다.저와 같은 문제로 질문 주신 커뮤니티의 다른 분께 프론트 파일과 주소를 남겨달라고 하셔서메일로 프론트 파일과 주소를 보내드린 상태입니다.한 번만 확인해 주신다면 감사드립니다.

'FastAPI에서 StaticFiles 클래스를 이용한 정적 파일 다루기' 수업 끝에 url_for 다음시간에 알려주신다고 영상이 끝나는데, 해당 강의영상은 어디에서 볼 수 있을까요?  ps. 수업 너무 알차게 잘 듣고 있습니다 :)

안녕하세요. fastapi를 사용해서 서버를 개발하고있는데, 한가지 불편한것이 있어서 여쭤보려고 합니다. 다름이 아니라 외부에 작성해놓은 함수를 import 할 때 생기는 문제인데요.project| - src| - | - api| - | - extension| - | - exception| - | - main.py이렇게 프로젝트 트리가 구성되어 있다고 했을때, project 경로에서 uvicorn src.main:app 으로 서버를 실행하면 ModuleNotFoundError: No module named 'extension' 이런 에러가 발생합니다. 이게 import 를 할 때 src.from extension.~ import ~ 이렇게 되어있지 않고 import 할 때, 자동으로 from extension.~ import ~ 이렇게 import가 되어서 모듈을 찾지 못해 발생하는 에러인 것 같은데요.혹시 자동으로 임포트 할 때부터 src.from extension.~ import ~ 이렇게 소스루트부터 import 하게 하는 설정이 따로 있을까요? 하나하나 적어주기가 너무 불편해서 여쭤봅니다 ㅠ

swagger 로 opt 생성, 검증 api 실행을 했더니 sqlalchemy.engine.Engine ROLLBACK 로그가 출력이 됩니다. 원인을 모르겠습니다...로그INFO: Application startup complete. INFO: 127.0.0.1:63654 - "GET /docs HTTP/1.1" 200 OK email-validator not installed, email fields will be treated as str. To install, run: pip install email-validator INFO: 127.0.0.1:63654 - "GET /openapi.json HTTP/1.1" 200 OK 2024-10-31 14:25:27,367 INFO sqlalchemy.engine.Engine SELECT DATABASE() 2024-10-31 14:25:27,367 INFO sqlalchemy.engine.Engine [raw sql] {} 2024-10-31 14:25:27,369 INFO sqlalchemy.engine.Engine SELECT @@sql_mode 2024-10-31 14:25:27,369 INFO sqlalchemy.engine.Engine [raw sql] {} 2024-10-31 14:25:27,369 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names 2024-10-31 14:25:27,370 INFO sqlalchemy.engine.Engine [raw sql] {} 2024-10-31 14:25:27,372 INFO sqlalchemy.engine.Engine BEGIN (implicit) 2024-10-31 14:25:27,377 INFO sqlalchemy.engine.Engine SELECT user.id, user.username, user.password, todo_1.id AS id_1, todo_1.contents, todo_1.is_done, todo_1.user_id FROM user LEFT OUTER JOIN todo AS todo_1 ON user.id = todo_1.user_id WHERE user.username = %(username_1)s 2024-10-31 14:25:27,377 INFO sqlalchemy.engine.Engine [generated in 0.00018s] {'username_1': 'admin'} INFO: 127.0.0.1:63658 - "POST /users/log-in HTTP/1.1" 200 OK 2024-10-31 14:25:27,603 INFO sqlalchemy.engine.Engine ROLLBACK INFO: 127.0.0.1:63664 - "POST /users/email/otp HTTP/1.1" 200 OK 2024-10-31 14:26:02,514 INFO sqlalchemy.engine.Engine BEGIN (implicit) 2024-10-31 14:26:02,514 INFO sqlalchemy.engine.Engine SELECT user.id, user.username, user.password, todo_1.id AS id_1, todo_1.contents, todo_1.is_done, todo_1.user_id FROM user LEFT OUTER JOIN todo AS todo_1 ON user.id = todo_1.user_id WHERE user.username = %(username_1)s 2024-10-31 14:26:02,514 INFO sqlalchemy.engine.Engine [cached since 35.14s ago] {'username_1': 'admin'} INFO: 127.0.0.1:63671 - "POST /users/email/verify HTTP/1.1" 200 OK Sending email to admin@fastapi.com! 2024-10-31 14:26:12,519 INFO sqlalchemy.engine.Engine ROLLBACK src/database/repository.pyfrom typing import List, Optional from fastapi import Depends from sqlalchemy import select, delete from sqlalchemy.orm import Session from database.connection import get_db from database.orm import ToDo, User class ToDoRepository: def __init__(self, session: Session = Depends(get_db)): self.session = session def get_todos(self) -> List[ToDo]: return list(self.session.scalars(select(ToDo))) def get_todo_by_todo_id(self, todo_id: int) -> ToDo | None: return self.session.scalar(select(ToDo).where(ToDo.id == todo_id)) def create_todo(self, todo: ToDo) -> ToDo: self.session.add(instance=todo) self.session.commit() self.session.refresh(instance=todo) return todo def update_todo(self, todo: ToDo) -> ToDo: self.session.add(instance=todo) self.session.commit() self.session.refresh(instance=todo) return todo def delete_todo(self, todo_id: int) -> None: self.session.execute(delete(ToDo).where(ToDo.id == todo_id)) self.session.commit() class UserRepository: def __init__(self, session: Session = Depends(get_db)): self.session = session def get_user_by_username(self, username: str) -> User | None: return self.session.scalar(select(User).where(User.username == username)) def save_user(self, user: User) -> User: self.session.add(instance=user) self.session.commit() self.session.refresh(instance=user) return user  src/service/user.pyimport random import time import bcrypt from datetime import datetime, timedelta from jose import jwt class UserService: encoding: str = "UTF-8" JWT_SECRET_KEY: str = "f002393019e8776398370aa671767b860b702854724591cd0da5fc97bda3daf1" JWT_ALGORITHM: str = "HS256" def hash_password(self, plain_password: str) -> str: hashed_password: bytes = bcrypt.hashpw( plain_password.encode(self.encoding), salt=bcrypt.gensalt() ) return hashed_password.decode(self.encoding) def verify_password( self, plain_password: str, hashed_password: str ) -> bool: return bcrypt.checkpw( plain_password.encode(self.encoding), hashed_password.encode(self.encoding) ) def creat_jwt(self, username: str) -> str: return jwt.encode( { "sub": username, "exp": datetime.now() + timedelta(days=1), }, self.JWT_SECRET_KEY, algorithm=self.JWT_ALGORITHM ) def decode_jwt(self, access_token: str) -> str: payload: dict = jwt.decode( access_token, self.JWT_SECRET_KEY, algorithms=[self.JWT_ALGORITHM] ) return payload["sub"] @staticmethod def create_otp() -> int: return random.randint(1000, 9999) @staticmethod def send_email_to_user(email: str) -> None: time.sleep(10) print(f"Sending email to {email}!") src/api/user.pyfrom fastapi import APIRouter, Depends, HTTPException, BackgroundTasks from cache import redis_client from database.orm import User from database.repository import UserRepository from schema.request import SignUpRequest, LoginRequest, CreateOTPRequest, VerifyOTPRequest from schema.response import UserSchema, JWTResponse from security import get_access_token from service.user import UserService router = APIRouter(prefix="/users", tags=["USER"]) @router.post("/sign-up", status_code=201) def user_sign_up_handler( request: SignUpRequest, user_service: UserService = Depends(), user_repository: UserRepository = Depends(), ): hashed_password: str = user_service.hash_password( plain_password=request.password ) user: User = User.create( username=request.username, hashed_password=hashed_password ) user: User = user_repository.save_user(user) return UserSchema.from_orm(user) @router.post("/log-in", status_code=200) def user_log_in_handler( request: LoginRequest, user_service: UserService = Depends(), user_repository: UserRepository = Depends(), ): user: User | None = user_repository.get_user_by_username( username=request.username ) if not user: raise HTTPException(status_code=404, detail="User Not Found") verified: bool = user_service.verify_password( plain_password=request.password, hashed_password=user.password ) if not verified: raise HTTPException(status_code=401, detail="Not Authorized") access_token: str = user_service.creat_jwt(username=user.username) return JWTResponse(access_token=access_token) @router.post("/email/otp") def create_otp_handler( request: CreateOTPRequest, _: str = Depends(get_access_token), user_service: UserService = Depends() ): otp: int = user_service.create_otp() redis_client.set(request.email, otp) redis_client.expire(request.email, 3 * 60) return {"otp": otp} @router.post("/email/verify") def verify_otp_handler( request: VerifyOTPRequest, background_tasks: BackgroundTasks, access_token: str = Depends(get_access_token), user_service: UserService = Depends(), user_repo: UserRepository = Depends(), ): otp: str | None = redis_client.get(request.email) if not otp: raise HTTPException(status_code=400, detail="Bad Request") if request.otp != int(otp): raise HTTPException(status_code=400, detail="Bad Request") username: str = user_service.decode_jwt(access_token=access_token) user: User | None = user_repo.get_user_by_username(username) if not user: raise HTTPException(status_code=404, detail="User Not Found") background_tasks.add_task( user_service.send_email_to_user, email="admin@fastapi.com" ) return UserSchema.from_orm(user)   

강사님 강좌를 여러 개 보면서 너무 만족을 하던 차라관심있던 fastapi의 목차를 자세히 보지 못하고 구매부터 했습니다.fastapi를 주로 api 서버로 많이 사용하는 것으로 알던 저는 jwt와 같은 토큰방식 로그인이 강좌에 있을 줄 알았는데, 없어서 많이 아쉬웠습니다.차후에 이러한 강좌들이 추가될 여지가 있을지 알고 싶습니다. 꼭 vue, react가 아닌, 자바스크립트를 이용한 간단한 통신방식의 프론트와의 연동으로도 구현이 될거 같은데.. 너무 아쉬운 마음에 요청 아닌 질문을 조심스레 드려 봅니다.