More research
Feed Stack Exchange Questions Scraper Data Into Your Pipeline
Feed Stack Exchange Questions Scraper Data Into Your Pipeline — a practical guide with a ready-to-run configuration and structured output.
Sep 16, 2026 · 11 min read · 2,697 words
Need to feed stack exchange questions scraper into your pipeline? Thirdwatch's Stack Exchange Questions Scraper turns it into structured rows — the fields you need, ready to export.
Skip the setup: Run this as a ready-to-go task on Apify — pre-loaded with the configuration from this guide.
Why Data-pipeline works
A recurring feed beats a one-off pull — wire the Actor once and let fresh rows land where your tools read them. This guide shows the configuration that pulls stack exchange questions scraper data from Stack Exchange Questions Scraper.
How does this compare to the alternatives?
Manual copy-paste works for a handful of rows; dedicated per-site scrapers on other platforms either gate the same data behind sign-in or charge flat monthly fees regardless of usage.
Why this Actor
- Purpose-built for Stack Exchange Questions Scraper — no generic crawling config to tune
- Pay-per-result pricing — you pay only for rows that come back
- Structured output ready for JSON/CSV export or API consumption
- No login, cookies, or API keys needed for the public surfaces it reads
How to do it in 3 steps
Step 1: Configure the input
Set the inputs as shown below — sites takes the targets, and the optional fields bound the run size.
Step 2: Run the Actor
Run it from the console, the API, or the linked saved task. One dataset row is written per record.
Step 3: Use the output
Every row carries the fields shown in the sample — export as JSON, CSV, or feed it straight into your pipeline.
{
"sites": [
"stackoverflow"
],
"searchQueries": [
"async python"
],
"tags": [
"python"
],
"sort": "activity",
"order": "desc",
"acceptedOnly": false,
"maxResultsPerSearch": 100
}Each dataset row looks like:
{
"question_id": 79999259,
"site": "stackoverflow",
"title": "How to resolve TypeError: isSubclass() arg 1 must be a class when issubclass() wasn't explicitly called?",
"body_text": "I am trying to resolve an issue regarding one of the files regarding a bot I'm working on. Whenever I attempt to load this specific gear which acts as extra functions related to levelling for my bot, I get the error TypeError: issubclass() arg 1 must be a class . However as much as I looked at threads with similar issues and looked through my code, I can't seem to find what it's referring to and no other gear has this issue so I figure the error must be somewhere in here that I missed. Related code is below for reference: progress.py #! usr/bin/env python\nfrom io import BytesIO\nfrom uuid import uuid7\n\nimport aiofiles\nimport stoat\nfrom PIL import Image, ImageDraw, ImageFile, ImageFont\nfrom stoat.ext import commands\n\nfrom utilities import ChaosBot, db\n\n# TODO: TypeError: issubclass() arg 1 must be a class\"\n\n\n# Create a gear for levelling\nclass Progress(commands.Gear, name=\"Progress\"):\n \"\"\"Commands about economy/levelling.\"\"\"\n\n GEAR_EMOJI = \"\ud83d\udcc8\"\n\n def __init__(self, bot: ChaosBot) -> None:\n self.bot = bot\n\n async def card_maker(self, ctx: commands.Context, user_id: int, server_id: int):\n # Get user information from ID\n with db.cursor() as cur:\n cur.execute(\n \"\"\"SELECT server_level, xp FROM levels WHERE member_id IN \n (SELECT member_id FROM members WHERE (server_id, user_id) = (%s, %s)) LIMIT 1\"\"\",\n (server_id, user_id),\n )\n level, xp = cur.fetchone()\n if self.bot.get_user(user_id):\n user = self.bot.get_user(user_id)\n username = user.display_name\n avatar_url = user.avatar.url()\n else:\n return await ctx.send(f\"Could not get user {user_id}!\")\n\n # Gather information for the level card\n threshold = (level + 1) * 25\n progress = (xp / threshold) * 870\n textcard = \"../assets/textcard.png\"\n levelcard = \"../assets/levelcard.png\"\n font = \"../assets/RobotoSlab-Regular.ttf\"\n result = \"../assets/result.png\"\n avatar_mask = \"../assets/avatar_mask.png\"\n bar_mask = \"../assets/bar_mask.png\"\n\n # Get the avatar of the target user from URL\n avatar_bytes = await self.bot.client.get_bytes(avatar_url)\n avatar: ImageFile.ImageFile = Image.open(BytesIO(avatar_bytes)).resize(\n (170, 170)\n )\n\n # Overlay the text card and avatar on the level card\n background: ImageFile.ImageFile = Image.open(levelcard)\n overlay: ImageFile.ImageFile = Image.open(textcard)\n background.paste(overlay, (200, 0), overlay)\n a_mask: ImageFile.ImageFile = (\n Image.open(avatar_mask).convert(\"L\").resize((170, 170))\n )\n background.paste(avatar, (15, 15), a_mask)\n\n # Print username, level, and xp on the level card\n nameFont = ImageFont.truetype(font, 40)\n subFont = ImageFont.truetype(font, 30)\n draw = ImageDraw.Draw(background)\n draw.text(\n (220, 20),\n username,\n font=nameFont,\n fill=\"white\",\n stroke_width=1,\n stroke_fill=(0, 0, 0),\n )\n draw.text(\n (220, 150),\n f\"Level - {level}\",\n font=subFont,\n fill=\"white\",\n stroke_width=1,\n stroke_fill=(0, 0, 0),\n )\n draw.text(\n (570, 150),\n f\"{xp}/{threshold} XP\",\n font=subFont,\n fill=\"white\",\n stroke_width=1,\n stroke_fill=(0, 0, 0),\n )\n\n # Draw progress bar on the level card\n img = Image.new(\"RGBA\", (870, 50), (0, 0, 0))\n draw = ImageDraw.Draw(img, \"RGBA\")\n draw.rounded_rectangle((0, 0, 870, 50), 25, fill=(255, 255, 255, 50))\n draw.rounded_rectangle((0, 0, progress, 50), 25, fill=(0, 128, 255))\n b_mask: ImageFile.ImageFile = Image.open(bar_mask).convert(\"L\")\n background.paste(img, (15, 225), b_mask)\n\n # Create and save the file and send it\n async with aiofiles.open(result, mode=\"wb\") as file:\n background.save(file, \"PNG\")\n await ctx.send(attachments=[stoat.Asset(filename=\"../assets/result.png\")])\n\n # Generates xp for a given message\n def give_xp(self, message: stoat.Message):\n words = message.content.split()\n if len(words) < 5:\n return 5\n else:\n return len(words)\n\n # Determines whether the user levels up or not; max level due to PostgreSQL int4 capacity\n # Capacity is 2147483647 and level 13107 is 2147254275\n def level_up(self, xp: int, level: int):\n threshold = (level + 1) * 25\n return xp >= threshold\n\n @commands.Gear.listener(\"on_message\")\n async def xp(self, message: stoat.Message):\n if message.author.bot:\n return\n author = message.author\n server = message.server\n channel = message.channel\n ctx = message.ctx\n person = message.author\n\n with db.cursor() as cur:\n cur.execute(\n \"\"\"SELECT server_level, xp FROM levels WHERE member_id IN \n (SELECT member_id FROM members WHERE (server_id, user_id) = (%s, %s)) LIMIT 1\"\"\",\n (server.id, author.id),\n )\n user = cur.fetchone()\n # If member is not registered, create an entry for them\n if not user:\n cur.execute(\n \"INSERT INTO levels (member_id, server_level, xp) VALUES (%s, %s, %s)\",\n (uuid7(), 0, 0),\n )\n # Increase user xp and level as necessary\n else:\n # Prevent users gaining more xp if they are already at the maximum level\n # For now, cap at 1000 but can go up to 13107 if needed\n level = user[0]\n if level > 999:\n return\n xp = user[1] + self.give_xp(message)\n if self.level_up(xp, level):\n level += 1\n xp = 0\n if ctx is None:\n await channel.send(\n f\"**{author.display_name}** reached level {level} on {server}!\"\n )\n else:\n await self.card_maker(self, ctx, person.id, message.server.id)\n cur.execute(\n \"UPDATE levels SET (server_level, xp) = (%s, %s) WHERE member_id = %s\",\n (level, xp, user[0]),\n )\n db.commit()\n\n @stoat.slash_command()\n async def level(\n self,\n ctx: commands.Context,\n person: stoat.Member | stoat.User | None,\n ):\n \"\"\"Check level of a person, defaults to checking your own level\"\"\"\n if not person:\n person = ctx.user\n with db.cursor() as cur:\n cur.execute(\n \"\"\"SELECT server_level, xp FROM levels WHERE member_id IN \n (SELECT member_id FROM members WHERE (server_id, user_id) = (%s, %s)) LIMIT 1\"\"\",\n (ctx.server.id, person.id),\n )\n record = cur.fetchone()\n\n # Return XP and level or nothing if user is not registered\n if not record:\n return await ctx.send(f\"{person.display_name} has no levels or XP!\")\n else:\n return await self.card_maker(self, ctx, person.id, ctx.server.id)\n\n @stoat.slash_command()\n async def leaderboard(self, ctx: commands.Context):\n \"\"\"Gets the top 10 highest ranked people on the server\"\"\"\n server = ctx.server\n # Sort the database for the highest 10 scoring on the server\n with db.cursor() as cur:\n cur.execute(\n \"\"\"SELECT l.server_level, l.xp, m.user_id FROM levels l\n INNER JOIN members m ON (m.member_id, m.server_id) = (l.member_id, %s)\n ORDER BY l.server_level DESC NULLS LAST, l.xp DESC NULLS LAST LIMIT 10\"\"\",\n [server.id],\n )\n leaders = cur.fetchall()\n embed_description = \"\"\n for position, leader in enumerate(leaders):\n # Get relevant information for each of the top 10\n level = leader[0]\n xp = leader[1]\n user_id = leader[2]\n user = self.bot.get_user(user_id) if self.bot.get_user(user_id) else user_id\n username = user.display_name if self.bot.get_user(user_id) else user_id\n threshold = (level + 1) * 25\n embed_description += (\n f\"{position + 1}. {username}\\tLevel: {level}\\t{xp}/{threshold} XP\\n\"\n )\n embed_description += f\"Requested by {ctx.author.display_name}\"\n embed = stoat.SendableEmbed(\n title=f\"{server.name} Leaderboard\",\n color=stoat.Colour.from_rgb(0, 128, 255),\n icon_url=ctx.server.icon.url(),\n )\n await ctx.send(embeds=[embed])\n\n\n# Add the gear to the bot\nasync def setup(bot: ChaosBot):\n await bot.add_gear(Progress(bot)) Edit: Narrowed the error down to this function but could use some help spotting it async def xp(self, message: stoat.Message):\n if message.author.bot:\n return\n author = message.author\n server = message.server\n channel = message.channel\n ctx = message.ctx\n person = message.author\n\n with db.cursor() as cur:\n cur.execute(\n \"\"\"SELECT server_level, xp FROM levels WHERE member_id IN \n (SELECT member_id FROM members WHERE (server_id, user_id) = (%s, %s)) LIMIT 1\"\"\",\n (server.id, author.id),\n )\n user = cur.fetchone()\n # If member is not registered, create an entry for them\n if not user:\n cur.execute(\n \"INSERT INTO levels (member_id, server_level, xp) VALUES (%s, %s, %s)\",\n (uuid7(), 0, 0),\n )\n # Increase user xp and level as necessary\n else:\n # Prevent users gaining more xp if they are already at the maximum level\n # For now, cap at 1000 but can go up to 13107 if needed\n level = user[0]\n if level > 999:\n return\n xp = user[1] + self.give_xp(message)\n if self.level_up(xp, level):\n level += 1\n xp = 0\n if ctx is None:\n await channel.send(\n f\"**{author.display_name}** reached level {level} on {server}!\"\n )\n else:\n await self.card_maker(self, ctx, person.id, message.server.id)\n cur.execute(\n \"UPDATE levels SET (server_level, xp) = (%s, %s) WHERE member_id = %s\",\n (level, xp, user[0]),\n )\n db.commit()```",
"body_html": "<p>I am trying to resolve an issue regarding one of the files regarding a bot I'm working on. Whenever I attempt to load this specific gear which acts as extra functions related to levelling for my bot, I get the error <code>TypeError: issubclass() arg 1 must be a class</code>. However as much as I looked at threads with similar issues and looked through my code, I can't seem to find what it's referring to and no other gear has this issue so I figure the error must be somewhere in here that I missed. Related code is below for reference:</p>\n<p>progress.py</p>\n<pre><code>#! usr/bin/env python\nfrom io import BytesIO\nfrom uuid import uuid7\n\nimport aiofiles\nimport stoat\nfrom PIL import Image, ImageDraw, ImageFile, ImageFont\nfrom stoat.ext import commands\n\nfrom utilities import ChaosBot, db\n\n# TODO: TypeError: issubclass() arg 1 must be a class"\n\n\n# Create a gear for levelling\nclass Progress(commands.Gear, name="Progress"):\n """Commands about economy/levelling."""\n\n GEAR_EMOJI = "\ud83d\udcc8"\n\n def __init__(self, bot: ChaosBot) -> None:\n self.bot = bot\n\n async def card_maker(self, ctx: commands.Context, user_id: int, server_id: int):\n # Get user information from ID\n with db.cursor() as cur:\n cur.execute(\n """SELECT server_level, xp FROM levels WHERE member_id IN \n (SELECT member_id FROM members WHERE (server_id, user_id) = (%s, %s)) LIMIT 1""",\n (server_id, user_id),\n )\n level, xp = cur.fetchone()\n if self.bot.get_user(user_id):\n user = self.bot.get_user(user_id)\n username = user.display_name\n avatar_url = user.avatar.url()\n else:\n return await ctx.send(f"Could not get user {user_id}!")\n\n # Gather information for the level card\n threshold = (level + 1) * 25\n progress = (xp / threshold) * 870\n textcard = "../assets/textcard.png"\n levelcard = "../assets/levelcard.png"\n font = "../assets/RobotoSlab-Regular.ttf"\n result = "../assets/result.png"\n avatar_mask = "../assets/avatar_mask.png"\n bar_mask = "../assets/bar_mask.png"\n\n # Get the avatar of the target user from URL\n avatar_bytes = await self.bot.client.get_bytes(avatar_url)\n avatar: ImageFile.ImageFile = Image.open(BytesIO(avatar_bytes)).resize(\n (170, 170)\n )\n\n # Overlay the text card and avatar on the level card\n background: ImageFile.ImageFile = Image.open(levelcard)\n overlay: ImageFile.ImageFile = Image.open(textcard)\n background.paste(overlay, (200, 0), overlay)\n a_mask: ImageFile.ImageFile = (\n Image.open(avatar_mask).convert("L").resize((170, 170))\n )\n background.paste(avatar, (15, 15), a_mask)\n\n # Print username, level, and xp on the level card\n nameFont = ImageFont.truetype(font, 40)\n subFont = ImageFont.truetype(font, 30)\n draw = ImageDraw.Draw(background)\n draw.text(\n (220, 20),\n username,\n font=nameFont,\n fill="white",\n stroke_width=1,\n stroke_fill=(0, 0, 0),\n )\n draw.text(\n (220, 150),\n f"Level - {level}",\n font=subFont,\n fill="white",\n stroke_width=1,\n stroke_fill=(0, 0, 0),\n )\n draw.text(\n (570, 150),\n f"{xp}/{threshold} XP",\n font=subFont,\n fill="white",\n stroke_width=1,\n stroke_fill=(0, 0, 0),\n )\n\n # Draw progress bar on the level card\n img = Image.new("RGBA", (870, 50), (0, 0, 0))\n draw = ImageDraw.Draw(img, "RGBA")\n draw.rounded_rectangle((0, 0, 870, 50), 25, fill=(255, 255, 255, 50))\n draw.rounded_rectangle((0, 0, progress, 50), 25, fill=(0, 128, 255))\n b_mask: ImageFile.ImageFile = Image.open(bar_mask).convert("L")\n background.paste(img, (15, 225), b_mask)\n\n # Create and save the file and send it\n async with aiofiles.open(result, mode="wb") as file:\n background.save(file, "PNG")\n await ctx.send(attachments=[stoat.Asset(filename="../assets/result.png")])\n\n # Generates xp for a given message\n def give_xp(self, message: stoat.Message):\n words = message.content.split()\n if len(words) < 5:\n return 5\n else:\n return len(words)\n\n # Determines whether the user levels up or not; max level due to PostgreSQL int4 capacity\n # Capacity is 2147483647 and level 13107 is 2147254275\n def level_up(self, xp: int, level: int):\n threshold = (level + 1) * 25\n return xp >= threshold\n\n @commands.Gear.listener("on_message")\n async def xp(self, message: stoat.Message):\n if message.author.bot:\n return\n author = message.author\n server = message.server\n channel = message.channel\n ctx = message.ctx\n person = message.author\n\n with db.cursor() as cur:\n cur.execute(\n """SELECT server_level, xp FROM levels WHERE member_id IN \n (SELECT member_id FROM members WHERE (server_id, user_id) = (%s, %s)) LIMIT 1""",\n (server.id, author.id),\n )\n user = cur.fetchone()\n # If member is not registered, create an entry for them\n if not user:\n cur.execute(\n "INSERT INTO levels (member_id, server_level, xp) VALUES (%s, %s, %s)",\n (uuid7(), 0, 0),\n )\n # Increase user xp and level as necessary\n else:\n # Prevent users gaining more xp if they are already at the maximum level\n # For now, cap at 1000 but can go up to 13107 if needed\n level = user[0]\n if level > 999:\n return\n xp = user[1] + self.give_xp(message)\n if self.level_up(xp, level):\n level += 1\n xp = 0\n if ctx is None:\n await channel.send(\n f"**{author.display_name}** reached level {level} on {server}!"\n )\n else:\n await self.card_maker(self, ctx, person.id, message.server.id)\n cur.execute(\n "UPDATE levels SET (server_level, xp) = (%s, %s) WHERE member_id = %s",\n (level, xp, user[0]),\n )\n db.commit()\n\n @stoat.slash_command()\n async def level(\n self,\n ctx: commands.Context,\n person: stoat.Member | stoat.User | None,\n ):\n """Check level of a person, defaults to checking your own level"""\n if not person:\n person = ctx.user\n with db.cursor() as cur:\n cur.execute(\n """SELECT server_level, xp FROM levels WHERE member_id IN \n (SELECT member_id FROM members WHERE (server_id, user_id) = (%s, %s)) LIMIT 1""",\n (ctx.server.id, person.id),\n )\n record = cur.fetchone()\n\n # Return XP and level or nothing if user is not registered\n if not record:\n return await ctx.send(f"{person.display_name} has no levels or XP!")\n else:\n return await self.card_maker(self, ctx, person.id, ctx.server.id)\n\n @stoat.slash_command()\n async def leaderboard(self, ctx: commands.Context):\n """Gets the top 10 highest ranked people on the server"""\n server = ctx.server\n # Sort the database for the highest 10 scoring on the server\n with db.cursor() as cur:\n cur.execute(\n """SELECT l.server_level, l.xp, m.user_id FROM levels l\n INNER JOIN members m ON (m.member_id, m.server_id) = (l.member_id, %s)\n ORDER BY l.server_level DESC NULLS LAST, l.xp DESC NULLS LAST LIMIT 10""",\n [server.id],\n )\n leaders = cur.fetchall()\n embed_description = ""\n for position, leader in enumerate(leaders):\n # Get relevant information for each of the top 10\n level = leader[0]\n xp = leader[1]\n user_id = leader[2]\n user = self.bot.get_user(user_id) if self.bot.get_user(user_id) else user_id\n username = user.display_name if self.bot.get_user(user_id) else user_id\n threshold = (level + 1) * 25\n embed_description += (\n f"{position + 1}. {username}\\tLevel: {level}\\t{xp}/{threshold} XP\\n"\n )\n embed_description += f"Requested by {ctx.author.display_name}"\n embed = stoat.SendableEmbed(\n title=f"{server.name} Leaderboard",\n color=stoat.Colour.from_rgb(0, 128, 255),\n icon_url=ctx.server.icon.url(),\n )\n await ctx.send(embeds=[embed])\n\n\n# Add the gear to the bot\nasync def setup(bot: ChaosBot):\n await bot.add_gear(Progress(bot))\n</code></pre>\n<p>Edit: Narrowed the error down to this function but could use some help spotting it</p>\n<pre><code> async def xp(self, message: stoat.Message):\n if message.author.bot:\n return\n author = message.author\n server = message.server\n channel = message.channel\n ctx = message.ctx\n person = message.author\n\n with db.cursor() as cur:\n cur.execute(\n """SELECT server_level, xp FROM levels WHERE member_id IN \n (SELECT member_id FROM members WHERE (server_id, user_id) = (%s, %s)) LIMIT 1""",\n (server.id, author.id),\n )\n user = cur.fetchone()\n # If member is not registered, create an entry for them\n if not user:\n cur.execute(\n "INSERT INTO levels (member_id, server_level, xp) VALUES (%s, %s, %s)",\n (uuid7(), 0, 0),\n )\n # Increase user xp and level as necessary\n else:\n # Prevent users gaining more xp if they are already at the maximum level\n # For now, cap at 1000 but can go up to 13107 if needed\n level = user[0]\n if level > 999:\n return\n xp = user[1] + self.give_xp(message)\n if self.level_up(xp, level):\n level += 1\n xp = 0\n if ctx is None:\n await channel.send(\n f"**{author.display_name}** reached level {level} on {server}!"\n )\n else:\n await self.card_maker(self, ctx, person.id, message.server.id)\n cur.execute(\n "UPDATE levels SET (server_level, xp) = (%s, %s) WHERE member_id = %s",\n (level, xp, user[0]),\n )\n db.commit()```\n</code></pre>\n",
"tags": [
"python",
"asynchronous",
"bots"
],
"score": 2,
"view_count": 164,
"answer_count": 1,
"is_answered": true,
"accepted_answer_id": 79999266,
"closed_date": null
}What to watch for
Rate the targets reasonably — large batches benefit from modest concurrency. Source sites may throttle heavy bursts; retry once and widen filters if a target surface is thin.
Related use cases
- Extract Stack Exchange Questions Scraper at Scale
- Monitor Stack Exchange Questions Scraper on a Schedule
- Compare Stack Exchange Questions Scraper Side by Side
- Export Stack Exchange Questions Scraper to CSV
- All Thirdwatch use-case guides
Run the Stack Exchange Questions Scraper on Apify Store — pay per result, free to try, no credit card to test.
Frequently asked questions
What does Stack Exchange Questions Scraper return?
One dataset row per result — with the fields shown in the sample output. Export as JSON, CSV, or Excel.
How do I control run size?
`sites` selects the targets and the max/limit fields bound how many rows come back. Start small, then scale.
Can I run this on a schedule?
Yes — save it as a task and attach a schedule in Apify Console for daily/weekly pulls.
Is this data public?
The Actor collects publicly visible data only — the same information a logged-out visitor sees.
What formats can I export?
JSON, CSV, Excel, XML, or direct API access to the dataset — plus webhooks and integrations.
What if a run returns fewer rows than expected?
The source limits some results; retry once, and widen the query or filters if the target surface is genuinely thin.