안녕하세요,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가 아닌, 자바스크립트를 이용한 간단한 통신방식의 프론트와의 연동으로도 구현이 될거 같은데.. 너무 아쉬운 마음에 요청 아닌 질문을 조심스레 드려 봅니다.

안녕하세요. 강의를 듣고 혼자 프로젝트 해보고있는데 왜인지 원인을 모르겠는 오류가 생겨서 여쭤봅니다.프로젝트를 uvicorn main:app --reload 명령어로 실행하면 무수히많은 에러 스택과 함께 아래와 같은 에러가 발생하는데요.pydantic.errors.PydanticUndefinedAnnotation: name 'Optional' is not definedPydantic에서 에러가 나는 것 같은데, 저는 Optional을 사용하지 않고, None으로 사용하고 있는데 있는데 왜 이러는걸까요?

안녕하세요, 먼저 좋은 내용의 강의 감사합니다. 다름 아니라 강의 비디오에서'FastAPI Form 다루기 - 01'부터 녹음 된 소리가 울림이 커서 명료하지 않게 들리는 것 같습니다. 혹시 번거로우시겠지만, 좋은 강의 경험을 위해 그 전까지의 영상과 같이 오디오를 깔끔하게 변환해주실 수 있을지 문의 드립니다. 감사합니다.

CORS 설명 주신 부분을 보면 처음 브라우저에서 request 날릴 때에는 Origin에 아무런 값이 들어가 있지 않아서 response를 받고 그 이후에 받은 html의 javascript 쪽에서 다른 요청을 보냈을 때 기존에 요청하면서 받았던 Origin 값이 같이 들어가면서 이슈가 발생한다고 말씀주셨는데, blog json data 테스트 하는 거 보면 origin이 null인데 바로 cors 에러가 나나요 ? 

pdf 14_async_sqlalchemy 데이터 검색(읽기) github 링크가 잘못 연결되어있습니다.https://gist.github.com/DaveLee-fun/3ff74108d3b58f8ffb731c1ec106691e연결되어 있는 페이지에 코드는 async 비동기 실행 코드가 아닌 그냥 sqlalchemy 코드입니다.

@app.get("/items") # 함수에 개별 인자값이 들어가 있는 경우 path parameter가 아닌 모든 인자는 query parameter # query parameter의 타입과 default값을 함수인자로 설정할 수 있음. async def read_item(skip: int = 0, limit: int = 2): return fake_items_db[skip : skip + limit] @app.get("/items_nd/") # 함수 인자값에 default 값이 주어지지 않으면 반드시 query parameter에 해당 인자가 주어져야 함. async def read_item_nd(skip: int, limit: int): return fake_items_db[skip : skip + limit]안녕하세요 url 차이에 관해 질문드립니다. 위에 url은 "/items" 인데, 아래 url은 "/items_nd/" 으로 끝에 /가 붙어있는데요. 둘다 query params로 구성되어 있는데 위는 /로 안끝나고, 아래는 왜 /로 끝났는지 이유가 궁금합니다.

섹션 12 SQLAlchemy 검색 기본 문법7분 54초@app.get("/users/{user_id}") def read_user(user_id: int, db: Session = Depends(get_db)): db_users = db.query(User.username, User.email).all() if db_users == []: return {"error": "User not found"} return {"db_users": db_users}강의 내용과 동일하게 작성하여 실행 했는데,아래와 같은 오류가 났습니다.구글 검색 해 봐도... 잘 모르겠어서 질문 남겨봅니다... ValueError: [ValueError('dictionary update sequence element #0 has length 6; 2 is required'), TypeError('vars() argument must have __dict__ attribute')]  

from fastapi import FastAPI app = FastAPI() @app.get('/') def heath_check_handler(): return {'ping': 'pong'} todo_data = { 1 : { 'id' : 1, 'contents' : '실전! FastAPI 섹션 0 수강', 'is_done' : True, }, 2: { 'id': 2, 'contents': '실전! FastAPI 섹션 1 수강', 'is_done': False, }, 3: { 'id': 3, 'contents': '실전! FastAPI 섹션 2 수강', 'is_done': False, }, } # 내림차순(큰값 -> 작은값) @app.get("/todos") def get_todos_handler(order: str | None = None): ret = list(todo_data.values()) if order and order == 'DESC': return ret[::-1] return ret위와 같이 강사님 코드 그대로 실행하고, 패키지 버전도 FastAPI==0.97.0인데, 쿼리 매개변수가 인식이 안되는데, 무슨 문제일까요?

동영상 음성이 너무 작습니다. 매번 따로 조절해줘야하는 번거로움이 있습니다.파이썬 문법보다는 실전적인 내용에 충실하시려는 의도는 느껴지지만 Type Hint, Decorator, Class 같은 고급문법 정도는 짚고 넘어가주시면 강의내용이 좀 더 이해가 잘 되지않을까 싶습니다.강의 초반부에 뒤에서 설명하시겠다고 하고 넘어가는 부분이 자주 있어서 강의 중반부부터 이해가 되지 않습니다. 실무적인 내용도 좋지만 최소한의 이론적인 설명도 명확하게 짚어주시면 어떨까 싶습니다.

강의 자료 다운로드 받고 압축 풀고 확인 해 봤는데,Fast API 개발환경설정 강의에서 보여주시는 00_FASTAPI_START 설명 파일(ppt자료?)이 없습니다.혹시 파일 누락된것이 아닐까요?확인 부탁드립니다.!

안녕하세요! 강의 잘 보고 있습니다. 강의를 보는 중에 헷갈리는 개념이 있어서 질문드려요!올려주신 강의 중 미들웨어 챕터 부분을 보고 있어요. 그런데 문득 미들웨어와 uvicorn, gunicorn, hypercorn 과 같은 WSGI, ASGI 서버 간의 관계에 대해서 궁금증이 들었습니다. 보통 언어에 상관없이 웹 통신을 한다라고 하면 아래와 같은 구조로 통신한다고 볼 수 있잖아요?(DB가 있다고 가정)"""Client(Web Browser) ---> Web Server ---> Web Application Server ---> DB"""그리고 파이썬에서는 uvicorn, gunicorn 과 같은 WSGI, ASGI 서버들이 Web Server 역할을, Flask, FastAPI 와 같은 프레임워크 서버들이 Web Application Server의 역할을 한다고 알고 있습니다. 그리고 저 Web Server 요소가 미들웨어라고 정의한다고 알고 있었는데요! 그런데 FastAPI 공식문서 상에서는 미들웨어라는 것을 소개할 때, "모든 요청 또는 응답에 대해서 사전 또는 사후 처리 로직을 추가할 때 사용하는 것" 이라고 소개를 하면서 문득 "그러면 여기서 이야기하는 미들웨어랑 uvicorn, gunicorn 과는 어떤 차이점이 있는거지?" 하면서 헷갈리더라구요. 자바에서는 미들웨어라는 개념이 JVM 위에서 동작하면서 DB 연결, TCP/IP 연결 등과 같이 다른 소프트웨어(ex.애플리케이션)가 잘 동작하도록 지원해주는 소프트웨어라고 알고 있는데요! 만약 이러한 관점에서 본다면 uvicorn, gunicorn 과 같은 WSGI 또는 ASGI 서버들도 미들웨어라고 볼 수는 없는 걸까요? ChatGPT 한테 물어보았을 때는 엄연히 다른 개념이라고 하는데.. 정확히 차이점을 잘 모르겠어서요. 답변 해주시면 감사하겠습니다! (참고로 저는 파이썬으로만 오랫동안 개발해와서 자바나 다른 언어 관련해서는 잘 모를 수도 있어서 다른 언어에 대한 비유로 답변 주시면 제가 잘 이해를 못할 수도 있습니다!)

세션을 실행하고 있어 curl로 실행하려고 하는데 아래와 같은 에러 메세지가 뜨네요.Invoke-WebRequest : 매개 변수 이름 'c'이(가) 모호하므로 매개 변수를 처리할 수 없습니다. 일치하는 항목은 다음과 같습니다. -Credential -CertificateThumbprint -Certificate -ContentType.위치 줄:1 문자:45+ curl -X POST http://127.0.0.1:8000/set/ -i -c cookies.txt+ ~~실행하면 위와 같은 에러메시지가 나오는데 어떻게 해결할 수 있나요? from fastapi import FastAPI,Request from fastapi.security import HTTPBasic from starlette.middleware.sessions import SessionMiddleware app=FastAPI() app.add_middleware(SessionMiddleware,secret_key="your-secret-key") @app.post("/set/") async def set_session(request:Request): request.session["username"]='smith' return {"message":"Session value"} @app.get("/get/") async def get_session(request:Request): username=request.session.get("username","Guest") return {"username":username}