[fastapi 심화] 파이썬FastApi & Firebase 커뮤니티 웹개발: 환경세팅 및 코드

이번 시간에는 파이썬 fastapi라이브러리를 이용하여 아래와같이 동작하는 간단한 커뮤니티 웹사이트를 만들어보겠습니다. 해당 프로젝트에서 사용하는 DB는 firestore database입니다.

 

웹사이트 동작화면

 

먼저 firebase 프로젝트를 생성해줍니다

firestore database생성 후 규칙을 다음과같이 변경해줍니다.

 

파이어스토어 규칙설정

 

 

그런 다음 비공개 키를 생성해서 후에 fastapi프로젝트의 루트디렉토리에 넣어줍니다.

비공개 키 생성

 

이제 fastapi 메인 코드를 작성해줍니다.

 

프로젝트 구조는 다음과 같습니다.

 

프로젝트 폴더 구조

 

model.py에는 해당 사이트에서 사용하는 db들의 key들로 구성한 데이터모델입니다.

해당 형태로 firestore database에 저장 및 조회됩니다.

 

[model.py]

# coding: utf-8

from sqlalchemy import Column, Integer, String, DateTime

from pydantic import BaseModel

import datetime

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class postTable(Base):

    __tablename__ = 'post'

    id = Column(Integer, primary_key=True, autoincrement=True)

    게시판종류이름 = Column(String(30), nullable=True)

    내용 = Column(String(100000), nullable=True)

    댓글갯수=Column(Integer, nullable=True)

    비밀번호 = Column(String(20), nullable=True)

    작성일시 = Column(String(20), nullable=True)

    작성자 = Column(String(20), nullable=True)

    작성자아이피 = Column(String(20), nullable=True)

    작성자아이피간단 = Column(String(20), nullable=True)

    제목 = Column(String(20), nullable=True)

    조회수 = Column(Integer, nullable=True)

class post(BaseModel):

    id: int

    게시판종류이름: str

    내용: str

    댓글갯수: int

    비밀번호: str

    작성일시: str

    작성자: str

    작성자아이피: str

    작성자아이피간단: str

    제목: str

    조회수: int

class commentTable(Base):

    __tablename__ = 'comment'

    id = Column(Integer, primary_key=True, autoincrement=True)

    postid = Column(Integer, nullable=True)

    내용 = Column(String(1000), nullable=True)

    비밀번호 = Column(String(30), nullable=True)

    작성일시 = Column(String(20), nullable=True)

    작성자=Column(String(30), nullable=True)

    작성자아이피 = Column(String(20), nullable=True)

    작성자아이피간단 = Column(String(20), nullable=True)

class comment(BaseModel):

    id: int

    postid: int

    내용: str

    비밀번호: str

    작성일시: str

    작성자: str

    작성자아이피: str

    작성자아이피간단: str

class community_typeTable(Base):

    __tablename__ = 'community_type'

    id = Column(Integer, primary_key=True, autoincrement=True)

    이름 = Column(String(30), nullable=True)

class community_type(BaseModel):

    id: int

    이름: str

if __name__ == "__main__":

    a=1

 

 

 

전에 생성한 비공개 키의 이름을 serviceAccountKey.json으로 바꿔 다음과같이 json파일 형태로 루트디렉토리 폴더에 넣어줍니다.

serviceAccountKey.json

 

serviceAccountKey.json파일 위치

 

 

다음은 핵심 코드인 mainlist.py입니다.

클라이언트 페이지인 html(javascript포함)에서 들어온 요청들을 받아서 서버에서 firestore DB 로부터 데이터를 읽고 쓰고 수정 및 삭제 행위들을 수행합니다.

 

 

[mainlist.py]

# https://github.com/Bipasori/ProjectCafeCOP/tree/main/FastAPI_MYSQL

from fastapi import FastAPI, Request

from fastapi.responses import HTMLResponse

from fastapi.templating import Jinja2Templates

from fastapi.staticfiles import StaticFiles

from typing import List

from starlette.middleware.cors import CORSMiddleware

from model import postTable, post, commentTable, comment

import datetime

import socket

import time

from threading import Thread

import threading

from fastapi import UploadFile, File

import os

import aiofiles

import base64

import uvicorn

import pyrebase

import json

import firebase_admin

from firebase_admin import firestore

from firebase_admin import credentials

import json

from google.cloud.firestore_v1 import FieldFilter

#서비스어카운트키 파일에 있는 json을 불러옴

if not firebase_admin._apps:

    cred = credentials.Certificate('serviceAccountKey.json')

    default_app = firebase_admin.initialize_app(cred)

#cred = credentials.Certificate("./serviceAccountKey.json")

#firebase_admin.initialize_app(cred)

db = firestore.client()

posts_ref = db.collection("posts")

comments_ref=db.collection("comments")

postkeys = postTable.__dict__.keys()

postkeylist = list(postkeys)

todeletepostkeylist = ['__module__', '__tablename__', '__doc__', '_sa_class_manager', '__table__', '__init__',

                   '__mapper__']

for i in range(len(todeletepostkeylist)):

    postkeylist.remove(todeletepostkeylist[i])

commentkeys = commentTable.__dict__.keys()

commentkeylist = list(commentkeys)

todeletecommentkeylist = ['__module__', '__tablename__', '__doc__', '_sa_class_manager', '__table__', '__init__',

                   '__mapper__']

for i in range(len(todeletecommentkeylist)):

    commentkeylist.remove(todeletecommentkeylist[i])

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

STATIC_DIR = os.path.join(BASE_DIR,'static/')

IMG_DIR = os.path.join(STATIC_DIR,'images/')

SERVER_IMG_DIR = os.path.join('http://localhost:80/','static/','images/')

curruser={"id":"","password":"","nickname":""}

templates = Jinja2Templates(directory="templates")

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

app.add_middleware(

    CORSMiddleware,

    allow_origins=["*"],

    allow_credentials=True,

    allow_methods=["*"],

    allow_headers=["*"],

)

def return_class_variables(A):

    return (A.__dict__)

# -------------------------------

# --------------홈---------------

# -------------------------------

@app.get("/",response_class=HTMLResponse)

async def root(request: Request):

    global curruser

    context = {}

    context['request'] =request

    context['curruser'] = curruser

    return templates.TemplateResponse("home.html", context)

# -------------------------------

# ----------------------------------------------------

# ----------글 추가수정삭제, 댓글 추가삭제------------

# ----------------------------------------------------

#글 목록 나열페이지

@app.get("/posts/{community_typestr}", response_class=HTMLResponse)

async def read_posts(request: Request, community_typestr: str):

    global postkeylist

    global posts_ref

    print("view_allposts >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

    context = {}

    #모든 글 정보들 firestore DB에서 가져오기(딕셔너리들의 리스트형태로 html페이지에 전달)

    postsdictlist=[]

    alldocs=posts_ref.stream()

    for doc in alldocs:

        docidstr=doc.id

        datadict=doc.to_dict()

        if datadict["게시판종류이름"]==community_typestr:

            if datadict not in postsdictlist:

                postsdictlist.append(datadict)

    postsdictlist.reverse()

    context['request'] = request

    context['postsdictlist'] = postsdictlist

    context['keylist']=postkeylist

    context['community_type']=community_typestr

    return templates.TemplateResponse("post_list.html", context)

#한 글 상세보기 페이지 (댓글 리스트도 표시해야함)

@app.get("/posts/{community_typestr}/{post_id}", response_class=HTMLResponse)

async def read_post(request: Request,community_typestr: str, post_id: int):

    global postkeylist

    global commentkeylist

    global posts_ref

    global comments_ref

    print("read_postdetail >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

    context = {}

    context['request'] = request

    context['post_id'] = post_id

    context['community_type']=community_typestr

    #이 글의 상세정보 firestore DB에서 가져오기

    postkeyvaluedict={}

    alldocs=posts_ref.stream()

    for doc in alldocs:

        docidstr=doc.id

        datadict=doc.to_dict()

        if datadict["id"]==post_id:

            postkeyvaluedict=datadict

            break

    #글 없으면 바로 리턴

    if len(postkeyvaluedict)==0:

        context['postexist']="no"

        return templates.TemplateResponse("post_detail.html", context)

    #이 글의 댓글들 정보 firestore DB에서 가져오기

    comments_keyvaluelists=[]

    last_comment={}

    alldocs=comments_ref.stream()

    for doc in alldocs:

        docidstr=doc.id

        datadict=doc.to_dict()

        if datadict["postid"]==post_id:

            if datadict not in comments_keyvaluelists:

                comments_keyvaluelists.append(datadict)

                last_comment=datadict

    context['postexist']="yes"

    context['keyvaluedict'] = postkeyvaluedict

    context['keylist'] = postkeylist

    context['commentkeylist']=commentkeylist

    context['comments_keyvaluelists'] = comments_keyvaluelists

    context['last_comment']=last_comment

    return templates.TemplateResponse("post_detail.html", context)

#글 추가/수정 리디렉트 페이지

@app.get("/go_postORedit/{community_typestr}/{createORedit}/{postid}", response_class=HTMLResponse)

async def go_createORedit_post(request: Request, community_typestr: str, createORedit:str, postid: int):

    global postkeylist

    global posts_ref

    print("view_allposts >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

    if createORedit=="create": #글 추가일떄

        context = {}

        #print(postkeylist)

        context['request'] = request

        context['keylist']=postkeylist

        context['community_type']=community_typestr

        context['createORedit']=createORedit

        context['postid']=postid

    elif createORedit=="edit": #글 수정일때

        context = {}

        postkeyvaluedict={}

        #선택된 글정보

        alldocs=posts_ref.stream()

        for doc in alldocs:

            docidstr=doc.id

            datadict=doc.to_dict()

            if datadict["id"]==post_id:

                postkeyvaluedict=datadict

                break

        context['request'] = request

        context['keyvaluedict'] = postkeyvaluedict

        context['keylist'] = postkeylist

        context['community_type']=community_typestr

        context['createORedit']=createORedit

        context['postid']=postid

    return templates.TemplateResponse("post_write_n_edit.html", context)

#글 추가

@app.post("/posts/{community_typestr}")

async def create_post(post: post, community_typestr: str):

    global posts_ref

    print("create_post >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

    #추가할 DB키 생성하기

    postlist = list(post)

    now = datetime.datetime.now()

    #post = postTable()

    alldbidlist=[]

    alldocs=posts_ref.stream()

    for doc in alldocs:

        docidstr=doc.id

        docdict=doc.to_dict()

        if docdict['id'] not in alldbidlist:

            alldbidlist.append(docdict['id'])

    alldbidlist.sort()

    lastdbid=0

    if len(alldbidlist)>=1:

        lastdbid=alldbidlist[len(alldbidlist)-1]

    string_number=str(lastdbid+1)

    dbkey = string_number.zfill(8)

    #추가할 데이터들 dict형태로 할당

    datadict={}

    for i in range(len(postlist)):

        #print(postlist[i])

        keyname=postlist[i][0]

        valuename=postlist[i][1]

        if keyname=='id':

            datadict[keyname]=int(string_number)

        else:

            datadict[keyname]=valuename

    #firestore DB에 추가

    posts_ref.document(dbkey).set(datadict)

    return { 'result_msg': f'{"put"} Registered...' }

#글 수정

@app.put("/posts/{community_typestr}")

async def modify_post(post: post, community_typestr: str,):

    global posts_ref

    print("modify_post >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

    postid = int(post.id)

    string_number=str(postid)

    dbkey = string_number.zfill(8)

    toupdate_posttitle=post.제목

    toupdate_postcontent=post.내용

    #firestore DB에 업데이트

    posts_ref.document(dbkey).update({"제목": toupdate_posttitle,"내용":toupdate_postcontent})

    return { 'result_msg': f"{str(postid)} updated..." }

#글 조회수 수정

@app.put("/posts_readcountUP/{community_typestr}")

async def modify_post(post: post, community_typestr: str):

    global posts_ref

    print("modify_post_readedcountup >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

    string_number=str(post.id)

    dbkey = string_number.zfill(8)

    #firestore DB에 업데이트

    posts_ref.document(dbkey).update({"조회수": post.조회수})

    return { 'result_msg': f"{str(post.id)} updated..." }

#댓글 추가 페이지

@app.post("/comments")

async def create_comment(comment: comment):

    global comments_ref

    global posts_ref

    print("create_comment >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

    commentlist = list(comment)

    #comment = commentTable()

    #댓글 키 생성

    alldbidlist=[]

    alldocs=comments_ref.stream()

    for doc in alldocs:

        docidstr=doc.id

        docdict=doc.to_dict()

        if docdict['id'] not in alldbidlist:

            alldbidlist.append(docdict['id'])

    alldbidlist.sort()

    lastdbid=0

    if len(alldbidlist)>=1:

        lastdbid=alldbidlist[len(alldbidlist)-1]

    string_number=str(lastdbid+1)

    dbkey = string_number.zfill(8)

    #해당 키로 댓글데이터 삽입 및 해당 글에 댓글갯수 추가

    datadict={}

    postid=0

    for i in range(len(commentlist)):

        keyname=commentlist[i][0]

        valuename=commentlist[i][1]

        if keyname=='id':

            datadict[keyname]=int(string_number)

        else:

            datadict[keyname]=valuename

            if keyname=='postid':

                postid=datadict[keyname]

    #firestore DB에 댓글 추가

    comments_ref.document(dbkey).set(datadict)

    #해당 글에 댓글갯수 +1

    string_number=str(postid)

    postdbkey = string_number.zfill(8)

    currcommentcount=0

    alldocs=posts_ref.stream()

    for doc in alldocs:

        docidstr=doc.id

        docdict=doc.to_dict()

        if docdict['id']==postid:

            currcommentcount=docdict['댓글갯수']

            break

    #댓글이 달린 글의 firestore DB에 댓글갯수 +1

    posts_ref.document(postdbkey).update({"댓글갯수": currcommentcount+1})

    return { 'result_msg': f'{"put"} Registered...' }

#비밀번호 확인페이지로 리디렉트

@app.get("/redirect_checkpassword_n_editordelete/{community_typestr}/{postorcomment}/{dbid}/{editordelete}/{postid}", response_class=HTMLResponse)

async def redirect_checkpassword_n_editordelete(request: Request, community_typestr: str, postorcomment:str, dbid: int, editordelete:str,postid:int):

    context={}

    password_errmsg=""

    if postorcomment=="post": #글일때

        context['postid']=dbid

        context['commentid']=-1

    elif postorcomment=="comment": #댓글일때

        context['commentid']=dbid

        context['postid']=postid

    context['request'] = request

    context['community_type']=community_typestr

    context['postorcomment']=postorcomment

    context['password_errmsg'] = password_errmsg

    context['editordelete']=editordelete

    return templates.TemplateResponse("pwcheckNeditordelete_postorcomment.html", context)

#비밀번호 확인후 글 수정/삭제 또는 댓글 삭제

@app.get("/checkpassword_n_editordelete/{community_typestr}/{postorcomment}/{dbid}/{editordelete}/{typed_password}", response_class=HTMLResponse)

async def checkpassword_n_editordelete(request: Request, community_typestr: str, postorcomment:str, dbid: int, editordelete:str,typed_password:str):

    global postkeylist

    global commentkeylist

    global posts_ref

    global comments_ref

    context={}

    password_errmsg=""

    realpassword=""

    context['request'] = request

    context['community_type']=community_typestr

    context['postorcomment']=postorcomment

    print("checkpassword_n_editordelete >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

    #글 수정 or 삭제

    if postorcomment=="post": #글일때

        context['postid']=dbid

        alldocs=posts_ref.stream()

        #해당 글 DB에서 가져와서 thispost_keyvaluedict 딕셔너리에 할당

        thispost_keyvaluedict={}

        for doc in alldocs:

            docidstr=doc.id

            datadict=doc.to_dict()

            if int(datadict["id"])==int(dbid):

                thispost_keyvaluedict=datadict

                break

        #진짜 비밀번호

        realpassword=thispost_keyvaluedict["비밀번호"]

        if editordelete=="edit": #글 수정

            #비밀번호가 다르면 에러메시지 할당후 리턴

            if typed_password!=realpassword:

                password_errmsg="비밀번호가 일치하지 않습니다"

                context['password_errmsg'] = password_errmsg

                context['commentid']=-1

                context['editordelete']=editordelete

                return templates.TemplateResponse("pwcheckNeditordelete_postorcomment.html", context)

            else: #비밀번호가 일치하면 글 수정페이지로 리턴

                context['keyvaluedict'] = thispost_keyvaluedict

                context['keylist'] = postkeylist

                context['createORedit']="edit"

                return templates.TemplateResponse("post_write_n_edit.html", context)

        elif editordelete=="delete": #글 삭제

            #비밀번호가 다르면 에러메시지 할당후 리턴

            if typed_password!=realpassword:

                password_errmsg="비밀번호가 일치하지 않습니다"

                context['password_errmsg'] = password_errmsg

                context['commentid']=-1

                context['editordelete']=editordelete

                return templates.TemplateResponse("pwcheckNeditordelete_postorcomment.html", context)

            else: #비밀번호가 일치하면 글 삭제 후 해당 게시판 목록으로 이동

                string_number=str(dbid)

                dbkey = string_number.zfill(8)

                #firestore DB에서 해당 글 삭제

                posts_ref.document(dbkey).delete()

                #해당 글에 달린 댓글들도 모두 DB에서 삭제하기

                allcomments=comments_ref.stream()

                for coment in allcomments:

                    commentdbkey=coment.id

                    commentdict=coment.to_dict()

                    if commentdict["postid"]==dbid:

                        commentid=commentdict["id"]

                        string_number=str(commentid)

                        commentdbkey = string_number.zfill(8)

                        #firestore DB에서 해당 댓글 삭제

                        comments_ref.document(commentdbkey).delete()

                postsdictlist=[]

                alldocs=posts_ref.stream()

                for doc in alldocs:

                    docidstr=doc.id

                    datadict=doc.to_dict()

                    if datadict["게시판종류이름"]==community_typestr:

                        if datadict not in postsdictlist:

                            postsdictlist.append(datadict)

                context['postsdictlist'] = postsdictlist

                context['keylist']=postkeylist

                return templates.TemplateResponse("post_list.html", context)

    #댓글 삭제

    elif postorcomment=="comment":

        #해당 댓글 DB에서 가져와서 thiscomment_keyvaluedict딕셔너리 할당

        thiscomment_keyvaluedict={}

        postid=0

        alldocs=comments_ref.stream()

        for doc in alldocs:

            docidstr=doc.id

            datadict=doc.to_dict()

            if int(datadict["id"])==int(dbid):

                postid=datadict["postid"]

                thiscomment_keyvaluedict=datadict

                break

        #해당 댓글의 postid

        context['postid']=postid

        #진짜 비밀번호

        realpassword=thiscomment_keyvaluedict["비밀번호"]

        #비밀번호가 다르면 에러메시지 할당후 리턴

        if typed_password!=realpassword:

            password_errmsg="비밀번호가 일치하지 않습니다"

            context['password_errmsg'] = password_errmsg

            context['commentid']=dbid

            context['editordelete']=editordelete

            return templates.TemplateResponse("pwcheckNeditordelete_postorcomment.html", context)

        else: #비밀번호가 일치하면 댓글 삭제 후 해당 글 상세 페이지로 이동

            #댓글 DB에서 삭제

            string_number=str(dbid)

            dbkey = string_number.zfill(8)

            #firestore DB에서 해당 댓글 삭제

            comments_ref.document(dbkey).delete()

            #해당 글에서 댓글갯수 -1

            string_number=str(postid)

            postdbkey = string_number.zfill(8)

            currcommentcount=0

            alldocs=posts_ref.stream()

            for doc in alldocs:

                docidstr=doc.id

                docdict=doc.to_dict()

                if docdict['id']==postid:

                    currcommentcount=docdict['댓글갯수']

                    break

            #댓글이 달린 글의 firestore DB에 댓글갯수 -1

            posts_ref.document(postdbkey).update({"댓글갯수": currcommentcount-1})

            context['post_id'] = postid

            #글 상세페이지로 리디렉트

            #이 글의 상세정보 firestore DB에서 가져오기

            postkeyvaluedict={}

            alldocs=posts_ref.stream()

            for doc in alldocs:

                docidstr=doc.id

                datadict=doc.to_dict()

                if datadict["id"]==postid:

                    postkeyvaluedict=datadict

                    break

            #글 없으면 바로 리턴

            if len(postkeyvaluedict)==0:

                context['postexist']="no"

                return templates.TemplateResponse("post_detail.html", context)

            #이 글의 댓글들 정보 firestore DB에서 가져오기

            comments_keyvaluelists=[]

            last_comment={}

            alldocs=comments_ref.stream()

            for doc in alldocs:

                docidstr=doc.id

                datadict=doc.to_dict()

                if datadict["postid"]==postid:

                    if datadict not in comments_keyvaluelists:

                        comments_keyvaluelists.append(datadict)

                        last_comment=datadict

            context['postexist']="yes"

            context['keyvaluedict'] = postkeyvaluedict

            context['keylist'] = postkeylist

            context['commentkeylist']=commentkeylist

            context['comments_keyvaluelists'] = comments_keyvaluelists

            context['last_comment']=last_comment

            return templates.TemplateResponse("post_detail.html", context)

if __name__ == "__main__":

    uvicorn.run(

        app='mainlist:app',

        host='0.0.0.0',

        port=91,

        reload=True

        #workers=16

    )

 

 

 

클라이언트 단의 html페이지들은 첨부한 전체fastpi-firestore커뮤니티소스코드.zip 파일에서 확인해주세요. (해당 소스코드에 serviceAccountKey.json는 빠져있으니 직접 생성한 파이어베이스 프로젝트에서 만든 serviceAccountKey.json파일을 넣어서 실행시켜주세요)

전체fastpi-firestore커뮤니티소스코드.zip

0.02MB

 

*해당 파이썬 프로그램을 실행시키려면 fastapi를 비롯해 각종 라이브러리들이 필요합니다. 

 

 

최종 완성 웹사이트 동작 모습입니다.

 

웹사이트 동작화면

 

해당 웹사이트는 아래 링크를 url에 치면 접속 가능합니다.

http://113.10.7.36:91/posts/자유