chore: replace aiosqlite with dataset and add bounded seen set to crawler

examples/crawler/Makefile

@@ -1,10 +1,10 @@
+# retoor <retoor@molodetz.nl>
 .PHONY: all env install run clean
 
 all: env install run
 
 env:
 	python3 -m venv .venv
-	.
 
 install:
 	. .venv/bin/activate && pip install -r requirements.txt

examples/crawler/README.md

@@ -1,34 +1,122 @@
-# Example Crawler Project
+# devRant Exhaustive Crawler
 
-This is a simple example crawler project. Follow the instructions below to set up and run the crawler.
+Author: retoor <retoor@molodetz.nl>
 
-## Setup
+An asynchronous crawler for comprehensive data collection from the devRant platform. Implements a producer-consumer architecture with multiple discovery strategies to maximize content coverage.
 
-1. Clone the repository or copy the project files to your local machine.
-2. Make sure you have Python 3 installed.
+## SSL Note
+
+The devRant API SSL certificate is expired. This crawler disables SSL verification to maintain connectivity. This is handled automatically by the API client.
+
+## Architecture
+
+The crawler employs four concurrent producers feeding into worker pools:
+
+| Producer | Strategy | Interval |
+|----------|----------|----------|
+| Recent | Paginate through recent rants | 2s |
+| Top | Paginate through top-rated rants | 5s |
+| Algo | Paginate through algorithm-sorted rants | 5s |
+| Search | Cycle through 48 programming-related search terms | 30s |
+
+Worker pools process discovered content:
+- 10 rant consumers fetch rant details and extract comments
+- 5 user consumers fetch profiles and discover associated rants
+
+Discovery graph: rants reveal users, users reveal more rants (from their profile, upvoted, favorites).
+
+## Data Storage
+
+Uses SQLite via the dataset library with:
+- Batched writes (100 items or 5s interval)
+- Automatic upsert for deduplication
+- Indexes on user_id, created_time, rant_id
+- State persistence for resume capability
+
+### Schema
+
+**rants**: id, user_id, text, score, created_time, num_comments, attached_image_url, tags, link, vote_state, user_username, user_score
+
+**comments**: id, rant_id, user_id, body, score, created_time, vote_state, user_username, user_score
+
+**users**: id, username, score, about, location, created_time, skills, github, website
+
+**crawler_state**: Persists producer positions (skip values, search term index)
 
 ## Usage
 
-1. Open a terminal in the project directory.
-2. Run `make` to set up the environment, install dependencies, and start the crawler:
+### Quick Start
 
 ```bash
 make
 ```
 
-This will create a virtual environment, install the package in editable mode from the parent directory, and run the main script.
+This creates a virtual environment, installs dependencies, and starts the crawler.
+
+### Manual Setup
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -e ../../.
+pip install -r requirements.txt
+python main.py
+```
+
+### Stopping
+
+Press `Ctrl+C` for graceful shutdown. The crawler will:
+1. Save current state to database
+2. Wait up to 30 seconds for queues to drain
+3. Flush remaining batched writes
+
+### Resuming
+
+Simply run again. The crawler loads saved state and continues from where it stopped.
+
+## Configuration
+
+Edit `main.py` to adjust:
+
+```python
+DB_FILE = "devrant.sqlite"
+CONCURRENT_RANT_CONSUMERS = 10
+CONCURRENT_USER_CONSUMERS = 5
+BATCH_SIZE = 100
+FLUSH_INTERVAL = 5.0
+```
+
+## Output
+
+The crawler logs statistics every 15 seconds:
+
+```
+[STATS] Rants Q'd/Proc: 1250/1200 | Users Q'd/Proc: 450/400 | Comments DB: 5600 | Queues (R/U): 50/50 | API Errors: 0
+```
 
 ## Cleanup
 
-To remove the virtual environment, run:
-
 ```bash
 make clean
 ```
 
-## Notes
+Removes the virtual environment. Database file (`devrant.sqlite`) is preserved.
 
-- The project installs the package with `-e ../../.` to include the parent package `devranta` in editable mode.
-- Ensure that the parent package is correctly set up in the directory structure.
+## Requirements
 
-Happy crawling!
+- Python 3.10+
+- dataset
+- aiohttp (via parent devranta package)
+
+## File Structure
+
+```
+crawler/
+├── main.py           # Entry point, configuration
+├── crawler.py        # Producer-consumer implementation
+├── database.py       # Dataset wrapper with batch queue
+├── requirements.txt  # Dependencies
+├── Makefile          # Build automation
+├── .venv/            # Virtual environment (created on first run)
+└── devrant.sqlite    # SQLite database (created on first run)
+```

examples/crawler/crawler.py

@@ -1,12 +1,50 @@
+# retoor <retoor@molodetz.nl>
 import asyncio
+import json
 import logging
+import time
+from collections import OrderedDict
 from typing import Set
 
 from database import DatabaseManager
-
 from devranta.api import Api, Rant
 
 
+SEARCH_TERMS = [
+    "python", "javascript", "java", "csharp", "golang", "rust",
+    "react", "angular", "vue", "node", "docker", "kubernetes",
+    "linux", "windows", "macos", "git", "github", "gitlab",
+    "sql", "mongodb", "redis", "api", "rest", "graphql",
+    "bug", "error", "crash", "debug", "fix", "issue",
+    "manager", "deadline", "meeting", "standup", "agile", "scrum",
+    "frontend", "backend", "fullstack", "devops", "cloud", "aws",
+    "typescript", "php", "ruby", "swift", "kotlin", "flutter",
+]
+
+
+class BoundedSeenSet:
+    def __init__(self, maxsize: int = 100000):
+        self._set = OrderedDict()
+        self._maxsize = maxsize
+
+    def add(self, item: int):
+        if item in self._set:
+            self._set.move_to_end(item)
+        else:
+            self._set[item] = None
+            if len(self._set) > self._maxsize:
+                self._set.popitem(last=False)
+
+    def __contains__(self, item: int) -> bool:
+        return item in self._set
+
+    def __len__(self) -> int:
+        return len(self._set)
+
+    def clear(self):
+        self._set.clear()
+
+
 class DevRantCrawler:
     def __init__(
         self, api: Api, db: DatabaseManager, rant_consumers: int, user_consumers: int
@@ -20,8 +58,14 @@ class DevRantCrawler:
         self.num_rant_consumers = rant_consumers
         self.num_user_consumers = user_consumers
 
-        self.seen_rant_ids: Set[int] = set()
-        self.seen_user_ids: Set[int] = set()
+        self.seen_rant_ids = BoundedSeenSet(maxsize=100000)
+        self.seen_user_ids = BoundedSeenSet(maxsize=100000)
+
+        self._recent_skip = 0
+        self._top_skip = 0
+        self._algo_skip = 0
+        self._search_term_index = 0
+
         self.stats = {
             "rants_processed": 0,
             "rants_added_to_db": 0,
@@ -74,29 +118,58 @@ class DevRantCrawler:
             f"Seeder finished: Queued {len(user_ids)} users to kickstart exploration."
         )
 
+    async def _save_state(self):
+        state = {
+            "recent_skip": self._recent_skip,
+            "top_skip": self._top_skip,
+            "algo_skip": self._algo_skip,
+            "search_term_index": self._search_term_index,
+            "last_saved": int(time.time()),
+        }
+        await self.db.save_crawler_state("producer_state", json.dumps(state))
+        logging.debug("Crawler state saved.")
+
+    async def _load_state(self):
+        state_json = await self.db.load_crawler_state("producer_state")
+        if state_json:
+            try:
+                state = json.loads(state_json)
+                self._recent_skip = state.get("recent_skip", 0)
+                self._top_skip = state.get("top_skip", 0)
+                self._algo_skip = state.get("algo_skip", 0)
+                self._search_term_index = state.get("search_term_index", 0)
+                logging.info(f"Loaded crawler state: {state}")
+            except json.JSONDecodeError:
+                logging.warning("Failed to decode crawler state, starting fresh.")
+
+    async def _state_saver(self):
+        logging.info("State saver started.")
+        while not self.shutdown_event.is_set():
+            await asyncio.sleep(60)
+            await self._save_state()
+
     async def _rant_producer(self):
-        logging.info("Rant producer started.")
-        skip = 0
+        logging.info("Recent rant producer started.")
         consecutive_empty_responses = 0
 
         while not self.shutdown_event.is_set():
             try:
-                logging.info(f"Producer: Fetching rants with skip={skip}...")
-                rants = await self.api.get_rants(sort="recent", limit=50, skip=skip)
+                logging.debug(f"Recent producer: Fetching rants with skip={self._recent_skip}...")
+                rants = await self.api.get_rants(sort="recent", limit=50, skip=self._recent_skip)
                 self.stats["producer_loops"] += 1
 
                 if not rants:
                     consecutive_empty_responses += 1
-                    logging.info(
-                        f"Producer: Feed returned empty. Consecutive empty hits: {consecutive_empty_responses}."
+                    logging.debug(
+                        f"Recent producer: Feed returned empty. Consecutive empty hits: {consecutive_empty_responses}."
                     )
                     if consecutive_empty_responses >= 5:
                         self.stats["end_of_feed_hits"] += 1
                         logging.info(
-                            "Producer: End of feed likely reached. Pausing for 15 minutes before reset."
+                            "Recent producer: End of feed likely reached. Pausing for 15 minutes before reset."
                         )
                         await asyncio.sleep(900)
-                        skip = 0
+                        self._recent_skip = 0
                         consecutive_empty_responses = 0
                     else:
                         await asyncio.sleep(10)
@@ -108,25 +181,98 @@ class DevRantCrawler:
                     await self._queue_rant_if_new(rant)
                     new_rants_found += 1
 
-                logging.info(
-                    f"Producer: Processed {new_rants_found} rants from feed. Total queued: {self.stats['rants_queued']}."
+                logging.debug(
+                    f"Recent producer: Processed {new_rants_found} rants from feed."
                 )
-                skip += len(rants)
+                self._recent_skip += len(rants)
                 await asyncio.sleep(2)
 
             except Exception as e:
-                logging.critical(
-                    f"Producer: Unhandled exception: {e}. Retrying in 60s."
+                logging.error(
+                    f"Recent producer: Unhandled exception: {e}. Retrying in 60s."
                 )
                 self.stats["api_errors"] += 1
                 await asyncio.sleep(60)
 
+    async def _top_rant_producer(self):
+        logging.info("Top rant producer started.")
+
+        while not self.shutdown_event.is_set():
+            try:
+                logging.debug(f"Top producer: Fetching rants with skip={self._top_skip}...")
+                rants = await self.api.get_rants(sort="top", limit=50, skip=self._top_skip)
+
+                if not rants:
+                    logging.info("Top producer: End of feed reached. Resetting after 1 hour.")
+                    self._top_skip = 0
+                    await asyncio.sleep(3600)
+                    continue
+
+                for rant in rants:
+                    await self._queue_rant_if_new(rant)
+
+                logging.debug(f"Top producer: Processed {len(rants)} rants.")
+                self._top_skip += len(rants)
+                await asyncio.sleep(5)
+
+            except Exception as e:
+                logging.error(f"Top producer: Unhandled exception: {e}. Retrying in 60s.")
+                self.stats["api_errors"] += 1
+                await asyncio.sleep(60)
+
+    async def _algo_rant_producer(self):
+        logging.info("Algo rant producer started.")
+
+        while not self.shutdown_event.is_set():
+            try:
+                logging.debug(f"Algo producer: Fetching rants with skip={self._algo_skip}...")
+                rants = await self.api.get_rants(sort="algo", limit=50, skip=self._algo_skip)
+
+                if not rants:
+                    logging.info("Algo producer: End of feed reached. Resetting after 1 hour.")
+                    self._algo_skip = 0
+                    await asyncio.sleep(3600)
+                    continue
+
+                for rant in rants:
+                    await self._queue_rant_if_new(rant)
+
+                logging.debug(f"Algo producer: Processed {len(rants)} rants.")
+                self._algo_skip += len(rants)
+                await asyncio.sleep(5)
+
+            except Exception as e:
+                logging.error(f"Algo producer: Unhandled exception: {e}. Retrying in 60s.")
+                self.stats["api_errors"] += 1
+                await asyncio.sleep(60)
+
+    async def _search_producer(self):
+        logging.info("Search producer started.")
+
+        while not self.shutdown_event.is_set():
+            try:
+                term = SEARCH_TERMS[self._search_term_index % len(SEARCH_TERMS)]
+                logging.debug(f"Search producer: Searching for '{term}'...")
+                rants = await self.api.search(term)
+
+                for rant in rants:
+                    await self._queue_rant_if_new(rant)
+
+                logging.debug(f"Search producer: Found {len(rants)} rants for '{term}'.")
+                self._search_term_index += 1
+                await asyncio.sleep(30)
+
+            except Exception as e:
+                logging.error(f"Search producer: Unhandled exception: {e}. Retrying in 60s.")
+                self.stats["api_errors"] += 1
+                await asyncio.sleep(60)
+
     async def _rant_consumer(self, worker_id: int):
         logging.info(f"Rant consumer #{worker_id} started.")
         while not self.shutdown_event.is_set():
             try:
-                rant_id = await self.rant_queue.get()
-                logging.info(
+                rant_id = await asyncio.wait_for(self.rant_queue.get(), timeout=5.0)
+                logging.debug(
                     f"Rant consumer #{worker_id}: Processing rant ID {rant_id}."
                 )
 
@@ -146,22 +292,27 @@ class DevRantCrawler:
                     self.stats["comments_added_to_db"] += 1
                     await self._queue_user_if_new(comment["user_id"])
 
-                logging.info(
+                logging.debug(
                     f"Rant consumer #{worker_id}: Finished processing rant {rant_id}, found {len(comments)} comments."
                 )
                 self.stats["rants_processed"] += 1
                 self.rant_queue.task_done()
 
+            except asyncio.TimeoutError:
+                continue
             except Exception as e:
                 logging.error(f"Rant consumer #{worker_id}: Unhandled exception: {e}")
-                self.rant_queue.task_done()
+                try:
+                    self.rant_queue.task_done()
+                except ValueError:
+                    pass
 
     async def _user_consumer(self, worker_id: int):
         logging.info(f"User consumer #{worker_id} started.")
         while not self.shutdown_event.is_set():
             try:
-                user_id = await self.user_queue.get()
-                logging.info(
+                user_id = await asyncio.wait_for(self.user_queue.get(), timeout=5.0)
+                logging.debug(
                     f"User consumer #{worker_id}: Processing user ID {user_id}."
                 )
 
@@ -183,14 +334,20 @@ class DevRantCrawler:
                         await self._queue_rant_if_new(rant_obj)
                         rants_found_on_profile += 1
 
-                logging.info(
+                logging.debug(
                     f"User consumer #{worker_id}: Finished user {user_id}, found and queued {rants_found_on_profile} associated rants."
                 )
                 self.stats["users_processed"] += 1
                 self.user_queue.task_done()
+
+            except asyncio.TimeoutError:
+                continue
             except Exception as e:
                 logging.error(f"User consumer #{worker_id}: Unhandled exception: {e}")
-                self.user_queue.task_done()
+                try:
+                    self.user_queue.task_done()
+                except ValueError:
+                    pass
 
     async def _stats_reporter(self):
         logging.info("Stats reporter started.")
@@ -206,13 +363,18 @@ class DevRantCrawler:
 
     async def run(self):
         logging.info("Exhaustive crawler starting...")
+        await self._load_state()
         await self._initial_seed()
 
         logging.info("Starting main producer and consumer tasks...")
         tasks = []
         try:
             tasks.append(asyncio.create_task(self._rant_producer()))
+            tasks.append(asyncio.create_task(self._top_rant_producer()))
+            tasks.append(asyncio.create_task(self._algo_rant_producer()))
+            tasks.append(asyncio.create_task(self._search_producer()))
             tasks.append(asyncio.create_task(self._stats_reporter()))
+            tasks.append(asyncio.create_task(self._state_saver()))
 
             for i in range(self.num_rant_consumers):
                 tasks.append(asyncio.create_task(self._rant_consumer(i + 1)))
@@ -232,6 +394,8 @@ class DevRantCrawler:
         logging.info("Shutting down... sending signal to all tasks.")
         self.shutdown_event.set()
 
+        await self._save_state()
+
         logging.info("Waiting for queues to empty... Press Ctrl+C again to force exit.")
         try:
             await asyncio.wait_for(self.rant_queue.join(), timeout=30)

examples/crawler/database.py

@@ -1,130 +1,226 @@
+# retoor <retoor@molodetz.nl>
+import asyncio
+import json
 import logging
-from typing import List
+from typing import Any, Dict, List, Optional
 
-import aiosqlite
+import dataset
 
 from devranta.api import Comment, Rant, UserProfile
 
 
 class DatabaseManager:
-    def __init__(self, db_path: str):
+    def __init__(self, db_path: str, batch_size: int = 100, flush_interval: float = 5.0):
         self.db_path = db_path
-        self._conn: aiosqlite.Connection | None = None
+        self.batch_size = batch_size
+        self.flush_interval = flush_interval
+        self._db: Optional[dataset.Database] = None
+        self._rant_batch: List[Dict[str, Any]] = []
+        self._comment_batch: List[Dict[str, Any]] = []
+        self._user_batch: List[Dict[str, Any]] = []
+        self._flush_task: Optional[asyncio.Task] = None
+        self._lock = asyncio.Lock()
 
     async def __aenter__(self):
         logging.info(f"Connecting to database at {self.db_path}...")
-        self._conn = await aiosqlite.connect(self.db_path)
-        await self._conn.execute("PRAGMA journal_mode=WAL;")
-        await self._conn.execute("PRAGMA foreign_keys=ON;")
-        await self.create_tables()
+        self._db = dataset.connect(
+            f"sqlite:///{self.db_path}?check_same_thread=False",
+            engine_kwargs={"connect_args": {"check_same_thread": False}}
+        )
+        await self._create_indexes()
+        self._flush_task = asyncio.create_task(self._periodic_flush())
         logging.info("Database connection successful.")
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        if self._conn:
-            await self._conn.close()
-            logging.info("Database connection closed.")
+        if self._flush_task:
+            self._flush_task.cancel()
+            try:
+                await self._flush_task
+            except asyncio.CancelledError:
+                pass
+        await self.flush_all()
+        if self._db:
+            self._db.close()
+        logging.info("Database connection closed.")
 
-    async def create_tables(self):
-        logging.info("Ensuring database tables exist...")
-        await self._conn.executescript(
-            """
-            CREATE TABLE IF NOT EXISTS users (
-                id INTEGER PRIMARY KEY,
-                username TEXT NOT NULL UNIQUE,
-                score INTEGER,
-                about TEXT,
-                location TEXT,
-                created_time INTEGER,
-                skills TEXT,
-                github TEXT,
-                website TEXT
-            );
-            CREATE TABLE IF NOT EXISTS rants (
-                id INTEGER PRIMARY KEY,
-                user_id INTEGER,
-                text TEXT,
-                score INTEGER,
-                created_time INTEGER,
-                num_comments INTEGER
-            );
-            CREATE TABLE IF NOT EXISTS comments (
-                id INTEGER PRIMARY KEY,
-                rant_id INTEGER,
-                user_id INTEGER,
-                body TEXT,
-                score INTEGER,
-                created_time INTEGER
-            );
-        """
-        )
-        await self._conn.commit()
-        logging.info("Table schema verified.")
+    async def _create_indexes(self):
+        def _sync_create():
+            self._db.query("CREATE INDEX IF NOT EXISTS idx_rants_user_id ON rants(user_id)")
+            self._db.query("CREATE INDEX IF NOT EXISTS idx_rants_created_time ON rants(created_time)")
+            self._db.query("CREATE INDEX IF NOT EXISTS idx_comments_rant_id ON comments(rant_id)")
+            self._db.query("CREATE INDEX IF NOT EXISTS idx_comments_user_id ON comments(user_id)")
+            self._db.query("CREATE INDEX IF NOT EXISTS idx_users_username ON users(username)")
+        await asyncio.to_thread(_sync_create)
+        logging.info("Database indexes verified.")
+
+    async def _periodic_flush(self):
+        while True:
+            await asyncio.sleep(self.flush_interval)
+            await self.flush_all()
+
+    async def flush_all(self):
+        async with self._lock:
+            await self._flush_rants()
+            await self._flush_comments()
+            await self._flush_users()
+
+    async def _flush_rants(self):
+        if not self._rant_batch:
+            return
+        batch = self._rant_batch.copy()
+        self._rant_batch.clear()
+
+        def _sync_insert():
+            table = self._db["rants"]
+            for rant in batch:
+                table.upsert(rant, ["id"])
+
+        await asyncio.to_thread(_sync_insert)
+        logging.debug(f"Flushed {len(batch)} rants to database")
+
+    async def _flush_comments(self):
+        if not self._comment_batch:
+            return
+        batch = self._comment_batch.copy()
+        self._comment_batch.clear()
+
+        def _sync_insert():
+            table = self._db["comments"]
+            for comment in batch:
+                table.upsert(comment, ["id"])
+
+        await asyncio.to_thread(_sync_insert)
+        logging.debug(f"Flushed {len(batch)} comments to database")
+
+    async def _flush_users(self):
+        if not self._user_batch:
+            return
+        batch = self._user_batch.copy()
+        self._user_batch.clear()
+
+        def _sync_insert():
+            table = self._db["users"]
+            for user in batch:
+                table.upsert(user, ["id"])
+
+        await asyncio.to_thread(_sync_insert)
+        logging.debug(f"Flushed {len(batch)} users to database")
+
+    def _transform_rant(self, rant: Rant) -> Dict[str, Any]:
+        attached_image = rant.get("attached_image")
+        image_url = None
+        if isinstance(attached_image, dict):
+            image_url = attached_image.get("url")
+        elif isinstance(attached_image, str):
+            image_url = attached_image
+
+        tags = rant.get("tags", [])
+        tags_str = json.dumps(tags) if tags else None
+
+        return {
+            "id": rant["id"],
+            "user_id": rant["user_id"],
+            "text": rant["text"],
+            "score": rant["score"],
+            "created_time": rant["created_time"],
+            "num_comments": rant["num_comments"],
+            "attached_image_url": image_url,
+            "tags": tags_str,
+            "link": rant.get("link"),
+            "vote_state": rant.get("vote_state"),
+            "user_username": rant.get("user_username"),
+            "user_score": rant.get("user_score"),
+        }
+
+    def _transform_comment(self, comment: Comment) -> Dict[str, Any]:
+        return {
+            "id": comment["id"],
+            "rant_id": comment["rant_id"],
+            "user_id": comment["user_id"],
+            "body": comment["body"],
+            "score": comment["score"],
+            "created_time": comment["created_time"],
+            "vote_state": comment.get("vote_state"),
+            "user_username": comment.get("user_username"),
+            "user_score": comment.get("user_score"),
+        }
+
+    def _transform_user(self, user: UserProfile, user_id: int) -> Dict[str, Any]:
+        return {
+            "id": user_id,
+            "username": user["username"],
+            "score": user["score"],
+            "about": user.get("about"),
+            "location": user.get("location"),
+            "created_time": user.get("created_time"),
+            "skills": user.get("skills"),
+            "github": user.get("github"),
+            "website": user.get("website"),
+        }
 
     async def add_rant(self, rant: Rant):
-        await self._conn.execute(
-            "INSERT OR IGNORE INTO rants (id, user_id, text, score, created_time, num_comments) VALUES (?, ?, ?, ?, ?, ?)",
-            (
-                rant["id"],
-                rant["user_id"],
-                rant["text"],
-                rant["score"],
-                rant["created_time"],
-                rant["num_comments"],
-            ),
-        )
-        await self._conn.commit()
+        async with self._lock:
+            self._rant_batch.append(self._transform_rant(rant))
+            if len(self._rant_batch) >= self.batch_size:
+                await self._flush_rants()
 
     async def add_comment(self, comment: Comment):
-        await self._conn.execute(
-            "INSERT OR IGNORE INTO comments (id, rant_id, user_id, body, score, created_time) VALUES (?, ?, ?, ?, ?, ?)",
-            (
-                comment["id"],
-                comment["rant_id"],
-                comment["user_id"],
-                comment["body"],
-                comment["score"],
-                comment["created_time"],
-            ),
-        )
-        await self._conn.commit()
+        async with self._lock:
+            self._comment_batch.append(self._transform_comment(comment))
+            if len(self._comment_batch) >= self.batch_size:
+                await self._flush_comments()
 
     async def add_user(self, user: UserProfile, user_id: int):
-        await self._conn.execute(
-            "INSERT OR IGNORE INTO users (id, username, score, about, location, created_time, skills, github, website) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
-            (
-                user_id,
-                user["username"],
-                user["score"],
-                user["about"],
-                user["location"],
-                user["created_time"],
-                user["skills"],
-                user["github"],
-                user["website"],
-            ),
-        )
-        await self._conn.commit()
+        async with self._lock:
+            self._user_batch.append(self._transform_user(user, user_id))
+            if len(self._user_batch) >= self.batch_size:
+                await self._flush_users()
 
     async def rant_exists(self, rant_id: int) -> bool:
-        async with self._conn.execute(
-            "SELECT 1 FROM rants WHERE id = ? LIMIT 1", (rant_id,)
-        ) as cursor:
-            return await cursor.fetchone() is not None
+        def _sync_check():
+            table = self._db["rants"]
+            return table.find_one(id=rant_id) is not None
+        return await asyncio.to_thread(_sync_check)
 
     async def user_exists(self, user_id: int) -> bool:
-        async with self._conn.execute(
-            "SELECT 1 FROM users WHERE id = ? LIMIT 1", (user_id,)
-        ) as cursor:
-            return await cursor.fetchone() is not None
+        def _sync_check():
+            table = self._db["users"]
+            return table.find_one(id=user_id) is not None
+        return await asyncio.to_thread(_sync_check)
 
     async def get_random_user_ids(self, limit: int) -> List[int]:
-        logging.info(
-            f"Fetching up to {limit} random user IDs from database for seeding..."
-        )
-        query = "SELECT id FROM users ORDER BY RANDOM() LIMIT ?"
-        async with self._conn.execute(query, (limit,)) as cursor:
-            rows = await cursor.fetchall()
-            user_ids = [row[0] for row in rows]
-            logging.info(f"Found {len(user_ids)} user IDs to seed.")
-            return user_ids
+        logging.info(f"Fetching up to {limit} random user IDs from database for seeding...")
+
+        def _sync_fetch():
+            result = self._db.query(f"SELECT id FROM users ORDER BY RANDOM() LIMIT {limit}")
+            return [row["id"] for row in result]
+
+        user_ids = await asyncio.to_thread(_sync_fetch)
+        logging.info(f"Found {len(user_ids)} user IDs to seed.")
+        return user_ids
+
+    async def get_all_rant_ids(self) -> List[int]:
+        def _sync_fetch():
+            result = self._db.query("SELECT id FROM rants")
+            return [row["id"] for row in result]
+        return await asyncio.to_thread(_sync_fetch)
+
+    async def get_all_user_ids(self) -> List[int]:
+        def _sync_fetch():
+            result = self._db.query("SELECT id FROM users")
+            return [row["id"] for row in result]
+        return await asyncio.to_thread(_sync_fetch)
+
+    async def save_crawler_state(self, key: str, value: str):
+        def _sync_save():
+            table = self._db["crawler_state"]
+            table.upsert({"key": key, "value": value}, ["key"])
+        await asyncio.to_thread(_sync_save)
+
+    async def load_crawler_state(self, key: str) -> Optional[str]:
+        def _sync_load():
+            table = self._db["crawler_state"]
+            row = table.find_one(key=key)
+            return row["value"] if row else None
+        return await asyncio.to_thread(_sync_load)

examples/crawler/main.py

@@ -1,45 +1,47 @@
-# main.py
+# retoor <retoor@molodetz.nl>
 import asyncio
 import logging
 import signal
 
 from crawler import DevRantCrawler
 from database import DatabaseManager
-
 from devranta.api import Api
 
-# --- Configuration ---
+
 DB_FILE = "devrant.sqlite"
-CONCURRENT_RANT_CONSUMERS = 10  # How many rants to process at once
-CONCURRENT_USER_CONSUMERS = 5  # How many user profiles to fetch at once
+CONCURRENT_RANT_CONSUMERS = 10
+CONCURRENT_USER_CONSUMERS = 5
+BATCH_SIZE = 100
+FLUSH_INTERVAL = 5.0
 
 
 async def main():
-    """Initializes and runs the crawler."""
     logging.basicConfig(
         level=logging.INFO,
         format="%(asctime)s [%(levelname)s] - %(message)s",
         datefmt="%Y-%m-%d %H:%M:%S",
     )
 
-    api = Api()
-
-    async with DatabaseManager(DB_FILE) as db:
-        crawler = DevRantCrawler(
-            api=api,
-            db=db,
-            rant_consumers=CONCURRENT_RANT_CONSUMERS,
-            user_consumers=CONCURRENT_USER_CONSUMERS,
-        )
-
-        # Set up a signal handler for graceful shutdown on Ctrl+C
-        loop = asyncio.get_running_loop()
-        for sig in (signal.SIGINT, signal.SIGTERM):
-            loop.add_signal_handler(
-                sig, lambda s=sig: asyncio.create_task(crawler.shutdown())
+    async with Api() as api:
+        async with DatabaseManager(
+            DB_FILE,
+            batch_size=BATCH_SIZE,
+            flush_interval=FLUSH_INTERVAL,
+        ) as db:
+            crawler = DevRantCrawler(
+                api=api,
+                db=db,
+                rant_consumers=CONCURRENT_RANT_CONSUMERS,
+                user_consumers=CONCURRENT_USER_CONSUMERS,
             )
 
-        await crawler.run()
+            loop = asyncio.get_running_loop()
+            for sig in (signal.SIGINT, signal.SIGTERM):
+                loop.add_signal_handler(
+                    sig, lambda s=sig: asyncio.create_task(crawler.shutdown())
+                )
+
+            await crawler.run()
 
 
 if __name__ == "__main__":

examples/crawler/requirements.txt

@@ -1 +1 @@
-aiosqlite
+dataset