import requests
import string
import random
import aiohttp
import inspect
class _AsyncMeta(type):
async def __call__(self, *args, **kwargs):
obb = object.__new__(self)
fn = obb.__init__(*args, **kwargs)
if inspect.isawaitable(fn):
await fn
return obb
[docs]class Client:
"""
Represents a synchronous Client for fetching reddit posts.
"""
def __init__(self, agent, key):
self.agent = agent
self.key = key
[docs] class Subreddit:
r"""
Gets the raw data on a subreddit, which can be fetched via the various methods here.
Parameters
----------
name : `str`
The name of the subreddit
mode : `str`
'top', 'hot', or 'new' kind of posts
limit : `int`
The number of posts to fetch, defaults to 25 if not specified
"""
def __init__(self, mode, name, limit: int = 25):
self.name = name
self.mode = mode
self.limit = limit
self.agent = ''.join(random.choices(
string.ascii_uppercase + string.digits, k=8))
if limit:
if mode == 'top':
self.json = requests.get(
'https://www.reddit.com/r/{}/top/.json?limit={}'.format(name, self.limit),
headers={'User-Agent': self.agent}).json()
elif mode == 'hot':
self.json = requests.get(
'https://www.reddit.com/r/{}/hot/.json?limit={}'.format(name, self.limit),
headers={'User-Agent': self.agent}).json()
elif mode == 'new':
self.json = requests.get(
'https://www.reddit.com/r/{}/new/.json{}'.format(name, self.limit),
headers={'User-Agent': self.agent}).json()
[docs] def selftext(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
selftext: `str`
The selftext of a post at the specified index.
"""
selftext = self.json['data']['children'][index]['data']['selftext']
return selftext
[docs] def title(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
title: `str`
The title of a post at the specified index.
"""
title = self.json['data']['children'][index]['data']['title']
return title
[docs] def post_url(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
post_url: `str`
The url of a post at the specified index.
"""
post_url = 'https://reddit.com' + self.json['data']['children'][index]['data']['permalink']
return post_url
[docs] def author(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
author: `str`
The author of a post at the specified index.
"""
author = self.json['data']['children'][index]['data']['author']
return author
[docs] def image(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
image: `str`
The image url in a post at the specified index.
Raises
------
KeyError
Raised if there's no image in the post.
"""
try:
image = self.json['data']['children'][index]['data']['url_overridden_by_dest']
return image
except KeyError:
raise KeyError('No image found')
[docs] def upvotes(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
upvotes: `str`
The number of upvotes in a post at the specified index.
"""
upvotes = self.json['data']['children'][index]['data']['ups']
return upvotes
[docs] def downvotes(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
downvotes: `str`
The number of downvotes in a post at the specified index.
"""
downvotes = self.json['data']['children'][index]['data']['downs']
return downvotes
[docs] def score(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
score: `str`
The score (upvotes-downvotes) of a post at the specified index.
"""
score = self.json['data']['children'][index]['data']['score']
return score
[docs] def permalink(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
permalink: `str`
The permalink of a post at the specified index.
"""
permalink = self.json['data']['children'][index]['data']['permalink']
return permalink
[docs] def raw_json(self):
r"""
Returns
-------
raw_json: `dict`
The raw json of the response.
"""
raw_json = self.json
return raw_json
[docs]class AsyncClient:
"""
Represents an asynchronous Client for fetching reddit posts.
"""
def __init__(self, agent, key):
self.agent = agent
self.key = key
[docs] class Subreddit(metaclass=_AsyncMeta):
r"""
Gets the raw data on a subreddit, which can be fetched via the various methods here.
Parameters
----------
name : `str`
The name of the subreddit
mode : `str`
'top', 'hot', or 'new' kind of posts
limit : `int`
The number of posts to fetch, defaults to 25 if not specified
"""
@staticmethod
async def _subreddit_main(mode, name, limit):
async with aiohttp.ClientSession() as session:
async with session.get(f"https://www.reddit.com/r/{name}/{mode}.json?limit={limit}") as resp:
return await resp.json()
async def __init__(self, mode, name, limit: int = 25):
self.name = name
self.mode = mode
self.limit = limit
self.agent = ''.join(random.choices(
string.ascii_uppercase + string.digits, k=8))
self.json = await self._subreddit_main(self.mode, self.name, self.limit)
[docs] async def selftext(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
selftext: `str`
The selftext of a post at the specified index.
"""
selftext = self.json['data']['children'][index]['data']['selftext']
return selftext
[docs] async def title(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
title: `str`
The title of a post at the specified index.
"""
title = self.json['data']['children'][index]['data']['title']
return title
[docs] async def post_url(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
post_url: `str`
The url of a post at the specified index.
"""
post_url = 'https://reddit.com' + self.json['data']['children'][index]['data']['permalink']
return post_url
[docs] async def author(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
author: `str`
The author of a post at the specified index.
"""
author = self.json['data']['children'][index]['data']['author']
return author
[docs] async def image(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
image: `str`
The image url in a post at the specified index.
Raises
-------
KeyError
If the post at the specified index does not have an image.
"""
try:
image = self.json['data']['children'][index]['data']['url_overridden_by_dest']
return image
except KeyError:
raise KeyError('No image found')
[docs] async def upvotes(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
upvotes: `str`
The number of upvotes in a post at the specified index.
"""
upvotes = self.json['data']['children'][index]['data']['ups']
return upvotes
[docs] async def downvotes(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
downvotes: `str`
The number of downvotes in a post at the specified index.
"""
downvotes = self.json['data']['children'][index]['data']['downs']
return downvotes
[docs] async def score(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
score: `str`
The score of a post (upvotes-downvotes) at the specified index.
"""
score = self.json['data']['children'][index]['data']['score']
return score
[docs] async def permalink(self, index: int):
r"""
Parameters
----------
index : `int`
The index of the post to fetch (first post is marked as 0)
Returns
-------
permalink: `str`
The permalink of a post at the specified index.
"""
permalink = self.json['data']['children'][index]['data']['permalink']
return permalink
[docs] async def raw_json(self):
r"""
Returns
-------
raw_json: `dict`
The raw json of the response.
"""
raw_json = self.json
return raw_json