목차
728x90
반응형
SMALL
📁 CRUD
CRUD : 데이터(엔티티) 처리를 위한 기초 연산
- Create
- Read
- Update
- Delete
📁 TCP 기반 PasteBin
PasteBin Service란 간단한 텍스트, 코드 등을 게시하는 서비스이다.
서버는 SET과 GET 메시지를 처리한다.
클라이언트는 SET과 GET 메시지를 전송한다.
프로토콜은 [SET|GET] KEY [VALUE]과 같고, utf-8로 인코딩된다.
- 'SET' KEY\r\nVALUE : KEY에 저장된 VALUE 저장
- 'GET' KEY : KEY에 저장된 VALUE 반환 (기본값 : "")
PasteBin 서비스의 서버와 클라이언트를 직접 구현해보자.
🌱 PasteBin Client 코드
import socket
FLAGS = _ = None
DEBUG = False
def main() :
if DEBUG :
print(f'Parsed arguments {FLAGS}')
print(f'Unparsed arguments {_}')
method = input('Input method : ').strip().upper()
key = input('Input key: ').strip()
message = f'{method} {key}\r\n'
if method == 'SET' :
value = input('Input value : ').strip()
message = f'{message}{value}'
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s :
s.connect((FLAGS.host, FLAGS.port))
s.sendall(message.encode('utf-8'))
data = s.recv(1500)
data = data.decode('utf-8')
print(f'Received : ')
print(f'{data}')
if __name__ == '__main__' :
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('--debug', action='store_true', help='The present debug message')
parser.add_argument('--host', default='127.0.0.1', type=str, help='IP address')
parser.add_argument('--port', default=8080, type=int, help='Port number')
FLAGS, _ = parser.parse_known_args()
DEBUG = FLAGS.debug
main()
🌱 PasteBin Server 코드
import socket
FLAGS = _ = None
DEBUG = False
def main():
if DEBUG:
print(f'Parsed arguments {FLAGS}')
print(f'Unparsed arguments {_}')
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((FLAGS.host, FLAGS.port))
s.listen(FLAGS.backlog)
print(f'Start server')
db = {}
while True:
try:
conn, addr = s.accept()
print(f'Connected: {addr=}')
with conn:
data = conn.recv(1500)
if DEBUG:
print(f'{data=}')
ptr = data.find('\r\n'.encode('utf-8'))
header = data[:ptr]
body = data[ptr:]
if DEBUG:
print(f'{header=}')
print(f'{body=}')
request = header.decode('utf-8')
if DEBUG:
print(f'{request=}')
chunks = request.split(' ')
method = chunks[0]
key = chunks[1]
print(f'{method=}')
print(f'{key=}')
value = 'Invalid request'
if method == 'GET':
value = db.get(key, '')
elif method == 'SET':
value = body.decode('utf-8')
db[key] = value
conn.sendall(value.encode('utf-8'))
except KeyboardInterrupt:
print('Shutdown server')
break
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('--debug', action='store_true',
help='The present debug message')
parser.add_argument('--host', default='0.0.0.0', type=str,
help='IP address')
parser.add_argument('--port', default=8080, type=int,
help='Port number')
parser.add_argument('--backlog', default=1, type=int,
help='Backlog')
FLAGS, _ = parser.parse_known_args()
DEBUG = FLAGS.debug
main()
위와 같은 결과를 얻을 수 있다.
📁 HTTP Method, RESTful API
CRUD를 HTTP Method로 표현하는 방법
- GET : Read의 역할
- PUT : Update의 역할
- POST : Create의 역할
- DELETE : Delete의 역할
REST는 Representational staet transfer의 줄임말로, 소프트웨어 아키텍쳐 스타일이다.
원격 컴퓨터(서버)에 대한 데이터 연산 (생성, 제공, 삭제)을 정의한다.
📁 FastAPI
RESTful API를 쉽게 만들 수 있는 파이썬 웹 프레임워크이다.
이외에도 기본적인 HTTP 앱 서버 제작 도구 Flask와 Django도 사용할 수 있다.
먼저 FastAPI를 설치해보자.
pip install fastapi
pip install uvicorn
# 한 번에 설치할 수 있는 명령
pip install fastapi uvicornpython이 설치되어있다는 가정 하에, pip 명령어를 사용해서 위와 같이 fastapi와 uvicorn을 설치할 수 있다.
🌱 Uvicorn
uvicorn이란 FastAPI로 웹 개발을 하기 위해 사용되는 HTTP Server이다.
FastAPI 프레임워크만으로는 웹 개발을 할 수 없다. FastAPI가 비동기 방식의 웹 서버 프레임워크이므로, 이와 호환 될 수 있는 비동기 방식의 웹 서버가 필요하다.
Uvicorn은 매우 가벼운 ASGI 서버로, ASGI(Asynchronous Server Gateway Interface)란 async, await 구문을 사용하는 비동기 방식 웹 서버를 의미한다. 비동기 방식이 가능한 파이썬 웹 서버 프레임워크와 어플리케이션 사이의 표준 인터페이스를 제공하므로 대표적으로 FastAPI와 함께 사용된다.
uvicorn main:app --reload --host 0.0.0.0 --port 8080위 명령어로 uvicorn 서버를 실행할 수 있다.
호스트 아이피 0.0.0.0과 포트번호 8080에서 main 파일을 실행하겠다는 의미이다.
GET, PUT, POST, DELETE를 처리하는 코드를 작성해볼건데, 해당 코드를 작성하는 파일 이름을 main.py라고 작성해야한다.
📁 RESTful API with SQLite3
SQLite3을 데이터베이스로 이용하여 PasteBin 서비스에서 RESTful API를 처리하는 코드를 작성해보자.
from fastapi import FastAPI
from pydantic import BaseModel
import sqlite3
app = FastAPI()
conn = sqlite3.connect('answer.db', check_same_thread=False)
cur = conn.cursor()
cur.execute('''CREATE TABLE IF NOT EXISTS Paste (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT);''')
conn.commit()
# Paste 클래스.
# 필드로 content를 가지고 있다.
class Paste(BaseModel):
content: str먼저 FastAPI와 SQLite3을 이용하기 위한 import가 필요하다.
🌱 GET 메소드 처리
# '/'에서 데이터를 가져오는 GET을 처리한다.
# 데이터베이스의 변화를 보기 위해 Paste 테이블의 정보를 가져온다.
@app.get('/')
def root():
res = cur.execute('''SELECT * FROM Paste;''')
data = res.fetchall()
return data
🌱 GET 파라미터 처리
# paste_id로 데이터를 가져오는 GET을 처리한다.
# 파라미터로 입력받은 paste_id가 id인 정보를 가져온다.
@app.get('/paste/{paste_id}')
def get_paste(paste_id: int):
res = cur.execute('''SELECT id, content
FROM Paste
WHERE id = ?''', (paste_id,))
data = res.fetchone()
conn.commit()
# 해당하는 정보가 존재하는 경우
# 가져온 정보의 id와 이를 이용해 만든 Paste 객체를 반환한다.
# 해당하는 정보가 존재하지 않는 경우 Paste 객체 자리에 None을 반환한다.
if data is not None:
paste = Paste(content=data[1])
return {'paste_id': data[0],
'paste': paste}
else:
return {'paste_id': paste_id,
'paste': None}
🌱 POST 처리
# content만 입력하여 등록하는 POST를 처리한다.
# content는 웹 내에서 request body 형태로 요청받고, 이를 Paste 테이블에 저장한다.
# 이때 id는 AUTOINCREMENT이므로 content만 저장한다.
@app.post('/paste/')
def post_paste(paste: Paste):
cur.execute('''INSERT INTO Paste (content)
VALUES (?);''', (paste.content,))
res = cur.execute('''SELECT last_insert_rowid();''')
data = res.fetchone()
conn.commit()
# POST한 정보의 id와 Paste 객체를 반환한다.
return {'paste_id': data[0],
'paste': paste}
🌱 PUT 처리
# paste_id를 파라미터로 받고, content를 입력받는 PUT을 처리한다.
@app.put('/paste/{paste_id}')
def put_paste(paste_id: int, paste: Paste):
# 현재 Paste 테이블에 paste_id를 id로 갖는 정보가 있는지 찾는다.
res = cur.execute('''SELECT COUNT(*) FROM Paste WHERE id = ?''', (paste_id,))
Isidin = res.fetchone()
# 이미 paste_id가 id인 정보가 있는 경우,
# 해당 정보를 입력받은 content로 UPDATE한다.
# 정보가 없는 경우, paste_id와 입력받은 content를 이용하여 INSERT한다.
if Isidin[0] != 0:
cur.execute('''UPDATE Paste
SET content = ?
WHERE id=?''', (paste.content, paste_id,))
else:
cur.execute('''INSERT INTO Paste (id, content)
VALUES (?,?);''', (paste_id, paste.content,))
conn.commit()
return {'paste_id': paste_id,
'paste': paste}
🌱 DELETE 처리
# paste_id를 파라미터로 받아 DELETE를 저장한다.
@app.delete('/paste/{paste_id}')
def delete_paste(paste_id:int):
# 현재 Paste 테이블에 paste_id를 id로 갖는 정보가 있는지 찾는다.
res = cur.execute('''SELECT COUNT(*) FROM Paste WHERE id = ?''', (paste_id,))
Isidin = res.fetchone()
# 이미 paste_id가 id인 정보가 있는 경우,
# paste_id를 id로 가진 정보를 DELETE한다.
# 정보가 없는 경우 DELETE를 실패했다고 반환한다.
if Isidin[0] != 0:
cur.execute('''DELETE FROM Paste
WHERE id = ?''', (paste_id,))
# 방금 DELETE한 정보를 다시 SELECT로 찾아서 반환함으로써
# null이 표시되는 것을 확인하여 정보가 지워졌음을 확실히함.
res = cur.execute('''SELECT id, content
FROM Paste
WHERE id = ?''', (paste_id,))
data = res.fetchone()
conn.commit()
return {'paste_id': paste_id,
'paste': data}
else:
return "failed to delete"
728x90
반응형
LIST
