81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
# -*- coding:utf-8 -*-
|
||
'''
|
||
@Author：Robin
|
||
@Email: 329080237@qq.com
|
||
@Wechat: 15618110227
|
||
@File: api.py
|
||
@Date: 2021/9/24 17:20
|
||
@Description:
|
||
'''
|
||
|
||
import base64
|
||
import requests, json
|
||
|
||
class RequestGit:
|
||
|
||
def __init__(self, request, api):
|
||
self.url = "http://39.103.231.107:3000/api/v1" + api
|
||
self.username = request.session.get("username", "")
|
||
self.password = request.session.get("password", "")
|
||
self.access_token = request.session.get("access_token")
|
||
self.headers = {
|
||
'Content-Type': 'application/json',
|
||
"Authorization": "Basic {}".format(self.getAuth()),
|
||
'Cookie': 'i_like_gitea=702577be6fb66a16; _csrf=0OFOibEHGY18vp0TNZhlblNuz6M6MTYzMjczMTg2NTk0MTc1MDI5NA',
|
||
}
|
||
|
||
#self.access_token = "7043f5103e78a74cd936d06da33dd47b33b2b05a"
|
||
|
||
|
||
def getAuth(self):
|
||
serect = self.username + ":" + self.password
|
||
auth = str(base64.b64encode(serect.encode("utf-8")), "utf-8")
|
||
return auth
|
||
|
||
|
||
def get(self, query=None, token=False):
|
||
if token:
|
||
request_url = self.url + "?access_token=" + self.access_token
|
||
else:
|
||
request_url = self.url
|
||
|
||
return requests.request("GET", request_url, headers=self.headers, params=query)
|
||
|
||
|
||
def post(self, payload=None, token=False):
|
||
# token = True 需要携带token访问，拼接成API URL
|
||
if token:
|
||
request_url = self.url + "?access_token=" + self.access_token
|
||
else:
|
||
request_url = self.url
|
||
|
||
return requests.request("POST", request_url, headers=self.headers, data=json.dumps(payload))
|
||
|
||
|
||
def put(self, payload=None, token=False):
|
||
# token = True 需要携带token访问，拼接成API URL
|
||
if token:
|
||
request_url = self.url + "?access_token=" + self.access_token
|
||
else:
|
||
request_url = self.url
|
||
|
||
return requests.request("PUT", request_url, headers=self.headers, data=json.dumps(payload))
|
||
|
||
|
||
def delete(self, payload=None, token=False):
|
||
if token:
|
||
request_url = self.url + "?access_token=" + self.access_token
|
||
else:
|
||
request_url = self.url
|
||
|
||
return requests.request("DELETE", request_url, headers=self.headers, data=json.dumps(payload))
|
||
|
||
def patch(self, payload=None, token=False):
|
||
# token = True 需要携带token访问，拼接成API URL
|
||
if token:
|
||
request_url = self.url + "?access_token=" + self.access_token
|
||
else:
|
||
request_url = self.url
|
||
|
||
return requests.request("PATCH", request_url, headers=self.headers, data=json.dumps(payload))
|