REQUESTS Archives - icodebroker

[PYTHON/REQUESTS] Session 클래스 : OLLAMA 서버 통신하기

■ Session 클래스를 사용해 OLLAMA 서버와 통신하는 방법을 보여준다. ▶ main.py


import json
import requests

from typing import Any

class OllamaClient:
    def __init__(self, baseURL : str = "http://localhost:11434"):
        self.baseURL = baseURL.rstrip('/')
        self.session = requests.Session()

    def generate(self, model : str, prompt : str,  stream : bool = False, **keywardArgumentDictionary) -> Any:
        url = f"{self.baseURL}/api/generate"

        data = {
            "model"  : model,
            "prompt" : prompt,
            "stream" : stream,
            **keywardArgumentDictionary
        }

        try:
            if stream:
                return self.session.post(url, json = data, stream = True)
            else:
                response = self.session.post(url, json=data)
                response.raise_for_status()
                return response.json()
        except requests.exceptions.RequestException as exception:
            print(f"API 요청 중 오류 발생 : {exception}")
            raise

def processStreamResponse(response):
    for line in response.iter_lines():
        if line:
            try:
                yield json.loads(line)
            except json.JSONDecodeError as exception:
                print(f"JSON 파싱 오류 : {exception}")
                continue

def main():
    ollamaClient = OllamaClient()

    model  = "bnksys/eeve-yanolja-v1:latest"
    prompt = "여기어때와 야놀자의 차이점을 알려주세요."

    try:
        print("=== 일반 응답 ===")
        response1 = ollamaClient.generate(model = model, prompt = prompt, stream = False)
        print(f"질문 : {prompt}")
        print(f"응답 : {response1['response']}\n")

        print("=== 스트리밍 응답 ===")
        print(f"질문 : {prompt}")
        print( "응답 :", end = " ", flush = True)

        response2 = ollamaClient.generate(model = model, prompt = prompt, stream = True)
        for chunkDictionary in processStreamResponse(response2):
            if chunkDictionary.get("response"):
                print(chunkDictionary["response"], end = "", flush = True)
        print()
    except Exception as exception:
        print(f"오류 발생 : {exception}")

if __name__ == "__main__":
    main()

import json

import requests

from typing import Any

class OllamaClient:

def __init__(self, baseURL : str = "http://localhost:11434"):

self.baseURL = baseURL.rstrip('/')

self.session = requests.Session()

def generate(self, model : str, prompt : str, stream : bool = False, **keywardArgumentDictionary) -> Any:

url = f"{self.baseURL}/api/generate"

data = {

"model" : model,

"prompt" : prompt,

"stream" : stream,

**keywardArgumentDictionary

}

try:

if stream:

return self.session.post(url, json = data, stream = True)

else:

response = self.session.post(url, json=data)

response.raise_for_status()

return response.json()

except requests.exceptions.RequestException as exception:

print(f"API 요청 중 오류 발생 : {exception}")

raise

def processStreamResponse(response):

for line in response.iter_lines():

if line:

try:

yield json.loads(line)

except json.JSONDecodeError as exception:

print(f"JSON 파싱 오류 : {exception}")

continue

def main():

ollamaClient = OllamaClient()

model = "bnksys/eeve-yanolja-v1:latest"

prompt = "여기어때와 야놀자의 차이점을 알려주세요."

try:

print("=== 일반 응답 ===")

response1 = ollamaClient.generate(model = model, prompt = prompt, stream = False)

print(f"질문 : {prompt}")

print(f"응답 : {response1['response']}\n")

print("=== 스트리밍 응답 ===")

print(f"질문 : {prompt}")

print( "응답 :", end = " ", flush = True)

response2 = ollamaClient.generate(model = model, prompt = prompt, stream = True)

for chunkDictionary in processStreamResponse(response2):

if chunkDictionary.get("response"):

print(chunkDictionary["response"], end = "", flush = True)

print()

except Exception as exception:

print(f"오류 발생 : {exception}")

if __name__ == "__main__":

main()

※ pip install requests 명령을 실행했다.

[PYTHON/REQUESTS] Retry 클래스 : HTTP 요청 오류 발생시 재시도하기

■ Retry 클래스를 사용해 HTTP 요청 오류 발생시 재시도하는 방법을 보여준다. ※ Retry 클래스의 생성자에서 total 인자는 총 재시도 카운트로 connect 인자와

[PYTHON/REQUESTS] HTTPAdapter 클래스 : 생성자에서 max_retries 인자를 사용해 HTTP 요청 오류 발생시 재시도하기

■ HTTPAdapter 클래스의 생성자에서 max_retries 인자를 사용해 HTTP 요청 오류 발생시 재시도하는 방법을 보여준다. ▶ main.py


from requests.adapters import HTTPAdapter
from requests.sessions import Session

def printResponse(title, response):
    print(title                                                                       )
    print(f"    response                         : {response}"                        )
    print(f"    response.url                     : {response.url}"                    )
    print(f"    response.headers['content-type'] : {response.headers['content-type']}")
    print(f"    response.encoding                : {response.encoding}"               )
    print(f"    response.status_code             : {response.status_code}"            )
    print()

def executeHTTPRequest(method, url, retryCount = 3):
    with Session() as session:
        httpAdapter = HTTPAdapter(max_retries = retryCount)

        session.mount("http://" , httpAdapter)
        session.mount("https://", httpAdapter)

        return session.request(method = method, url = url)

response = executeHTTPRequest("get", "https://icodebroker.com")

printResponse("테스트", response)

"""
테스트
    response                         : <Response [200]>
    response.url                     : https://icodebroker.com/
    response.headers['content-type'] : text/html; charset=UTF-8
    response.encoding                : UTF-8
    response.status_code             : 200
"""

from requests.adapters import HTTPAdapter

from requests.sessions import Session

def printResponse(title, response):

print(title )

print(f" response : {response}" )

print(f" response.url : {response.url}" )

print(f" response.headers['content-type'] : {response.headers['content-type']}")

print(f" response.encoding : {response.encoding}" )

print(f" response.status_code : {response.status_code}" )

print()

def executeHTTPRequest(method, url, retryCount = 3):

with Session() as session:

httpAdapter = HTTPAdapter(max_retries = retryCount)

session.mount("http://" , httpAdapter)

session.mount("https://", httpAdapter)

return session.request(method = method, url = url)

response = executeHTTPRequest("get", "https://icodebroker.com")

printResponse("테스트", response)

"""

테스트

response : <Response [200]>

response.url : https://icodebroker.com/

response.headers['content-type'] : text/html; charset=UTF-8

response.encoding : UTF-8

response.status_code : 200

"""

▶ requirements.txt


certifi==2024.6.2
charset-normalizer==3.3.2
idna==3.7
requests==2.32.3
urllib3==2.2.2

certifi==2024.6.2

charset-normalizer==3.3.2

idna==3.7

requests==2.32.3

urllib3==2.2.2

※ pip

[PYTHON/COMMON] requests 패키지 설치하기

■ requests 패키지를 설치하는 방법을 보여준다. 1. 명령 프롬프트를 실행한다. 2. 명령 프롬프트에서 아래 명령을 실행한다. ▶ 실행 명령


pip install requests

pip install requests

[PYTHON/REQUESTS] get 함수 : timeout 인자를 사용해 타임아웃 시간 설정하기

■ get 함수의 timeout 인자를 사용해 타임아웃 시간 설정하는 방법으로 보여준다. ▶ 예제 코드 (PY)


import requests

def printResponse(title, response):
    print(title                                                                       )
    print(f"    response                         : {response}"                        )
    print(f"    response.url                     : {response.url}"                    )
    print(f"    response.headers['content-type'] : {response.headers['content-type']}")
    print(f"    response.encoding                : {response.encoding}"               )
    print(f"    response.status_code             : {response.status_code}"            )
   #print(f"    response.text                    : {response.text}"                   )
    print()

response = requests.get(url = "http://google.com", timeout = 1) # 1초 후 타임아웃 에러가 발생한다.

printResponse("requests.get 호출 결과", response)

"""
requests.get 호출 결과
    response                         : <Response [200]>
    response.url                     : http://www.google.com/
    response.headers['content-type'] : text/html; charset=ISO-8859-1
    response.encoding                : ISO-8859-1
    response.status_code             : 200
"""

import requests

def printResponse(title, response):

print(title )

print(f" response : {response}" )

print(f" response.url : {response.url}" )

print(f" response.headers['content-type'] : {response.headers['content-type']}")

print(f" response.encoding : {response.encoding}" )

print(f" response.status_code : {response.status_code}" )

#print(f" response.text : {response.text}" )

print()

response = requests.get(url = "http://google.com", timeout = 1) # 1초 후 타임아웃 에러가 발생한다.

printResponse("requests.get 호출 결과", response)

"""

requests.get 호출 결과

response : <Response [200]>

response.url : http://www.google.com/

response.headers['content-type'] : text/html; charset=ISO-8859-1

response.encoding : ISO-8859-1

response.status_code : 200

"""

▶ requirements.txt


certifi==2024.6.2
charset-normalizer==3.3.2
idna==3.7
requests==2.32.3
urllib3==2.2.1

certifi==2024.6.2

charset-normalizer==3.3.2

idna==3.7

requests==2.32.3

urllib3==2.2.1

[PYTHON/REQUESTS] get 함수 : HTTP GET 요청하기

■ get 함수를 사용해 HTTP GET 요청을 처리하는 방법으로 보여준다. ▶ 예제 코드 (PY)


import requests

def printResponse(title, response):
    print(title                                                                       )
    print(f"    response                         : {response}"                        )
    print(f"    response.url                     : {response.url}"                    )
    print(f"    response.headers['content-type'] : {response.headers['content-type']}")
    print(f"    response.encoding                : {response.encoding}"               )
    print(f"    response.status_code             : {response.status_code}"            )
   #print(f"    response.text                    : {response.text}"                   )
    print()

response = requests.get(url = "http://google.com")

printResponse("requests.get 호출 결과", response)

"""
requests.get 호출 결과
    response                         : <Response [200]>
    response.url                     : http://www.google.com/
    response.headers['content-type'] : text/html; charset=ISO-8859-1
    response.encoding                : ISO-8859-1
    response.status_code             : 200
"""

import requests

def printResponse(title, response):

print(title )

print(f" response : {response}" )

print(f" response.url : {response.url}" )

print(f" response.headers['content-type'] : {response.headers['content-type']}")

print(f" response.encoding : {response.encoding}" )

print(f" response.status_code : {response.status_code}" )

#print(f" response.text : {response.text}" )

print()

response = requests.get(url = "http://google.com")

printResponse("requests.get 호출 결과", response)

"""

requests.get 호출 결과

response : <Response [200]>

response.url : http://www.google.com/

response.headers['content-type'] : text/html; charset=ISO-8859-1

response.encoding : ISO-8859-1

response.status_code : 200

"""

▶ requirements.txt


certifi==2024.6.2
charset-normalizer==3.3.2
idna==3.7
requests==2.32.3
urllib3==2.2.1

certifi==2024.6.2

charset-normalizer==3.3.2

idna==3.7

requests==2.32.3

urllib3==2.2.1