-
Notifications
You must be signed in to change notification settings - Fork 1
/
shopify_blog_manager.py
187 lines (148 loc) · 10.5 KB
/
shopify_blog_manager.py
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
import json
import requests
from tenacity import (
retry, wait_random_exponential, stop_after_attempt, retry_if_exception_type
)
store_url = "https://example.myshopify.com"
access_token = "your_access_token"
blog_id = "your_blog_id" # Replace with the actual blog ID
headers = {
"X-Shopify-Access-Token": access_token,
"Content-Type": "application/json"
}
@retry(wait=wait_random_exponential(multiplier=1, min=4, max=10), stop=stop_after_attempt(10), retry=retry_if_exception_type(requests.exceptions.RequestException))
def get_blogs(store_url, access_token):
"""
Retrieves all blogs from the specified Shopify store using the Shopify Admin API.
The function makes an HTTP GET request to the Shopify API's blog endpoint. It is decorated with a retry mechanism which retries the request on encountering specific exceptions (like network-related errors) with exponential backoff.
Parameters:
- store_url (str): The base URL of the Shopify store, e.g., 'https://example.myshopify.com'.
- access_token (str): The access token used for authenticating with the Shopify API.
Returns:
- dict: A dictionary object containing the response data with all blogs if successful.
Raises:
- Exception: If the request fails or the response status is not 200, it raises an exception with the error code and text.
Usage:
- blogs = get_blogs('https://example.myshopify.com', 'your_access_token')
print(blogs) # Prints the list of blogs retrieved from the Shopify store.
"""
headers = {"X-Shopify-Access-Token": access_token, "Content-Type": "application/json"}
response = requests.get(f"{store_url}/admin/api/2023-10/blogs.json", headers=headers)
if response.status_code == 200:
return response.json() # Returns the list of blogs
else:
raise Exception(f"Error: {response.status_code}, {response.text}") # Raises an exception if something goes wrong
@retry(wait=wait_random_exponential(multiplier=1, min=4, max=10), stop=stop_after_attempt(10), retry=retry_if_exception_type(requests.exceptions.RequestException))
def create_blog_with_metafields(store_url, access_token, title, key, value, type, namespace):
"""
Creates a new blog with specified metafields in the Shopify store using the Shopify Admin API.
This function makes an HTTP POST request to the Shopify API's blog creation endpoint. It includes metafields in the request, allowing for additional custom data to be associated with the blog. The function automatically retries on encountering specific exceptions (like network-related errors) with exponential backoff.
Parameters:
- store_url (str): The base URL of the Shopify store, e.g., 'https://example.myshopify.com'.
- access_token (str): The access token used for authenticating with the Shopify API.
- title (str): Title of the new blog.
- key (str): Key for the metafield.
- value (str): Value for the metafield.
- type (str): Type of the metafield (e.g., 'string', 'integer').
- namespace (str): Namespace for the metafield, used for grouping and identifying metafields.
Returns:
- dict: A dictionary object containing the response data with details of the created blog if successful.
Raises:
- Exception: If the request fails or the response status is not 201, it raises an exception with the error code and text.
Usage:
- new_blog = create_blog_with_metafields('https://example.myshopify.com', 'your_access_token', 'My New Blog', 'sponsor', 'Shopify', 'string', 'global')
print(new_blog) # Prints the details of the newly created blog.
"""
pass
@retry(wait=wait_random_exponential(multiplier=1, min=4, max=10), stop=stop_after_attempt(10), retry=retry_if_exception_type(requests.exceptions.RequestException))
def create_article(store_url, access_token, blog_id, title, author, tags, body_html):
"""
Creates a new article in a specified blog on the Shopify store using the Shopify Admin API.
This function makes an HTTP POST request to the Shopify API's article creation endpoint for a specific blog. It includes details such as the article's title, author, tags, and body. The function automatically retries on encountering specific exceptions (like network-related errors) with exponential backoff.
Parameters:
- store_url (str): The base URL of the Shopify store, e.g., 'https://example.myshopify.com'.
- access_token (str): The access token used for authenticating with the Shopify API.
- blog_id (str): The ID of the blog where the article will be created.
- title (str): Title of the new article.
- author (str): Author of the new article.
- tags (str): Comma-separated tags for the new article.
- body_html (str): HTML content of the new article.
Returns:
- dict: A dictionary object containing the response data with details of the created article if successful.
Raises:
- Exception: If the request fails or the response status is not 201, it raises an exception with the error code and text.
Usage:
- new_article = create_article('https://example.myshopify.com', 'your_access_token', '123456789', 'My First Article', 'John Doe', 'Fashion, Trends', '<p>This is the body of my article.</p>')
print(new_article) # Prints the details of the newly created article.
"""
pass
@retry(wait=wait_random_exponential(multiplier=1, min=4, max=10), stop=stop_after_attempt(10), retry=retry_if_exception_type(requests.exceptions.RequestException))
def get_articles(store_url, access_token, blog_id):
"""
Retrieves all articles from a specified blog in the Shopify store using the Shopify Admin API.
This function makes an HTTP GET request to the Shopify API's articles endpoint for a specific blog. It is decorated with a retry mechanism which retries the request on encountering specific exceptions (like network-related errors) with exponential backoff.
Parameters:
- store_url (str): The base URL of the Shopify store, e.g., 'https://example.myshopify.com'.
- access_token (str): The access token used for authenticating with the Shopify API.
- blog_id (str): The ID of the blog from which to retrieve articles.
Returns:
- dict: A dictionary object containing the response data with details of all articles if successful.
Raises:
- Exception: If the request fails or the response status is not 200, it raises an exception with the error code and text.
Usage:
- articles = get_articles('https://example.myshopify.com', 'your_access_token', '123456789')
print(articles) # Prints the list of articles retrieved from the specified blog.
"""
headers = {"X-Shopify-Access-Token": access_token, "Content-Type": "application/json"}
response = requests.get(f"{store_url}/admin/api/2023-10/blogs/{blog_id}/articles.json", headers=headers)
if response.status_code == 200:
return response.json() # Returns the list of articles
else:
raise Exception(f"Error: {response.status_code}, {response.text}") # Raises an exception if something goes wrong
@retry(wait=wait_random_exponential(multiplier=1, min=4, max=10), stop=stop_after_attempt(10), retry=retry_if_exception_type(requests.exceptions.RequestException))
def get_article_count(store_url, access_token, blog_id):
"""
Retrieves the count of articles in a specified blog on the Shopify store using the Shopify Admin API.
This function makes an HTTP GET request to the Shopify API's article count endpoint for a specific blog. It is decorated with a retry mechanism which retries the request on encountering specific exceptions (like network-related errors) with exponential backoff.
Parameters:
- store_url (str): The base URL of the Shopify store, e.g., 'https://example.myshopify.com'.
- access_token (str): The access token used for authenticating with the Shopify API.
- blog_id (str): The ID of the blog for which to retrieve the article count.
Returns:
- dict: A dictionary object containing the response data with the count of articles if successful.
Raises:
- Exception: If the request fails or the response status is not 200, it raises an exception with the error code and text.
Usage:
- article_count = get_article_count('https://example.myshopify.com', 'your_access_token', '123456789')
print(article_count) # Prints the count of articles in the specified blog.
"""
headers = {"X-Shopify-Access-Token": access_token, "Content-Type": "application/json"}
response = requests.get(f"{store_url}/admin/api/2023-10/blogs/{blog_id}/articles/count.json", headers=headers)
if response.status_code == 200:
return response.json() # Returns the count of articles
else:
raise Exception(f"Error: {response.status_code}, {response.text}") # Raises an exception if something goes wrong
@retry(wait=wait_random_exponential(multiplier=1, min=4, max=10), stop=stop_after_attempt(10), retry=retry_if_exception_type(requests.exceptions.RequestException))
def delete_article(store_url, access_token, blog_id, article_id):
"""
Deletes an article from a specified blog on the Shopify store using the Shopify Admin API.
This function makes an HTTP DELETE request to the Shopify API's article endpoint for a specific article in a blog. It is decorated with a retry mechanism which retries the request on encountering specific exceptions (like network-related errors) with exponential backoff.
Parameters:
- store_url (str): The base URL of the Shopify store, e.g., 'https://example.myshopify.com'.
- access_token (str): The access token used for authenticating with the Shopify API.
- blog_id (str): The ID of the blog from which the article will be deleted.
- article_id (str): The ID of the article to delete.
Returns:
- None: If the delete operation is successful.
Raises:
- Exception: If the request fails or the response status is not 200, it raises an exception with the error code and text.
Usage:
- delete_article('https://example.myshopify.com', 'your_access_token', '123456789', '987654321')
# Deletes the specified article from the specified blog.
"""
headers = {"X-Shopify-Access-Token": access_token, "Content-Type": "application/json"}
response = requests.delete(f"{store_url}/admin/api/2023-10/blogs/{blog_id}/articles/{article_id}.json", headers=headers)
if response.status_code == 200:
print("Article deleted successfully!") # Inform the user of success
else:
raise Exception(f"Error: {response.status_code}, {response.text}") # Raises an exception if something goes wrong