-
Notifications
You must be signed in to change notification settings - Fork 8
Expand file tree
/
Copy pathpython_wpapi.py
More file actions
156 lines (122 loc) · 5.64 KB
/
python_wpapi.py
File metadata and controls
156 lines (122 loc) · 5.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
# -*- coding: utf-8 -*-
import requests
from python_wpapi.errors import WpApiError
class WpApi():
def __init__(self, base_url, user=None, password=None, headers_base={}):
self.auth = None
if user:
self.auth = (user, password)
self.base_url = base_url.strip('/') + '/wp-json/wp/v2'
self.headers_base = headers_base
def join_headers(self, headers):
if self.headers_base:
return dict(self.headers_base, **headers)
return headers
def _request(self, endpoint, method='GET', files=None, headers={}, **kwargs):
params = {}
if method == 'GET':
params = kwargs
data = None
headers = {'Content-Length': '0'}
else:
data = kwargs
headers = self.join_headers(headers)
response = requests.request(
method,
endpoint,
auth=self.auth,
params=params,
json=data,
headers=headers,
files=files,
verify=False
)
if not response.status_code // 100 == 2:
error = WpApiError.factory(response)
raise error
return response.json()
def get_posts(self):
endpoint = '{}/posts'.format(self.base_url)
return self._request(endpoint)
def get_post(self, id, context='view'):
endpoint = '{}/posts/{}'.format(self.base_url, id)
return self._request(endpoint, context=context)
def create_post(self, **data):
endpoint = '{}/posts'.format(self.base_url)
return self._request(endpoint, method='POST', **data)
def update_post(self, id, **data):
endpoint = '{}/posts/{}'.format(self.base_url, id)
return self._request(endpoint, method='POST', **data)
def delete_post(self, id, force=False):
endpoint = '{}/posts/{}'.format(self.base_url, id)
return self._request(endpoint, method='DELETE', force=force)
def get_medias(self):
endpoint = '{}/media'.format(self.base_url)
return self._request(endpoint)
def get_media(self, id, context='view'):
endpoint = '{}/media/{}'.format(self.base_url, id)
return self._request(endpoint, context=context)
def create_media(self, filename, content_type, file_data, **data):
headers = {'Content-Disposition': 'attachment; filename="{}"'.format(filename)}
files = {'file': (filename, file_data, content_type, {'Expires': '0'})}
endpoint = '{}/media'.format(self.base_url)
return self._request(endpoint, method='POST',
headers=headers, files=files, **data)
def update_media(self, id, **data):
endpoint = '{}/media/{}'.format(self.base_url, id)
return self._request(endpoint, method='POST', **data)
def delete_media(self, id, force=False):
endpoint = '{}/media/{}'.format(self.base_url, id)
return self._request(endpoint, method='DELETE', force=force)
def get_users(self, parameters='?per_page=10'):
endpoint = '{}/users/{}'.format(self.base_url, parameters)
return self._request(endpoint)
def get_user(self, id, context='view'):
endpoint = '{}/users/{}'.format(self.base_url, id)
return self._request(endpoint, context=context)
def create_user(self, username, email, password, **data):
endpoint = '{}/users'.format(self.base_url)
return self._request(endpoint, method='POST',
username=username, email=email, password=password, **data)
def update_user(self, id, **data):
endpoint = '{}/users/{}'.format(self.base_url, id)
return self._request(endpoint, method='POST', **data)
def delete_user(self, id):
endpoint = '{}/users/{}'.format(self.base_url, id)
return self._request(endpoint, method='DELETE', force=True)
def get_taxonomies(self, type=None):
endpoint = '{}/taxonomies'.format(self.base_url)
return self._request(endpoint, type=type)
def get_taxonomy(self, slug):
endpoint = '{}/taxonomies/{}'.format(self.base_url, slug)
return self._request(endpoint)
def get_categories(self, parameters='?per_page=10'):
endpoint = '{}/categories/{}'.format(self.base_url, parameters)
return self._request(endpoint)
def get_category(self, id, context='view'):
endpoint = '{}/categories/{}'.format(self.base_url, id)
return self._request(endpoint, context=context)
def create_category(self, name, **data):
endpoint = '{}/categories'.format(self.base_url)
return self._request(endpoint, method='POST', name=name, **data)
def update_category(self, id, **data):
endpoint = '{}/categories/{}'.format(self.base_url, id)
return self._request(endpoint, method='POST', **data)
def delete_category(self, id):
endpoint = '{}/categories/{}'.format(self.base_url, id)
return self._request(endpoint, method='DELETE', force=True)
def get_tags(self):
endpoint = '{}/tags'.format(self.base_url)
return self._request(endpoint)
def get_tag(self, id, context='view'):
endpoint = '{}/tags/{}'.format(self.base_url, id)
return self._request(endpoint, context=context)
def create_tag(self, name, **data):
endpoint = '{}/tags'.format(self.base_url)
return self._request(endpoint, method='POST', name=name, **data)
def update_tag(self, id, **data):
endpoint = '{}/tags/{}'.format(self.base_url, id)
return self._request(endpoint, method='POST', **data)
def delete_tag(self, id):
endpoint = '{}/tags/{}'.format(self.base_url, id)
return self._request(endpoint, method='DELETE', force=True)