Initial PR to add support for stackoverflow teams

README.md

@@ -32,6 +32,7 @@ Coming Soon...
  - [X] Bookstack - by [@flifloo](https://github.com/flifloo) :pray:
  - [X] Mattermost - by [@itaykal](https://github.com/Itaykal) :pray:
  - [X] RocketChat - by [@flifloo](https://github.com/flifloo) :pray:
+ - [X] Stackoverflow Teams - by [@allen-munsch](https://github.com/allen-munsch) :pray:
  - [ ] Gitlab Issues (In PR :pray:)
  - [ ] Zendesk (In PR :pray:)
  - [ ] Azure DevOps (In PR :pray:)

app/api/data_source.py

@@ -85,7 +85,7 @@ async def list_locations(request: Request, data_source_name: str, config: dict)
 @router.post("")
 async def connect_data_source(request: Request, dto: AddDataSourceDto, background_tasks: BackgroundTasks) -> int:
  logger.info(f"Adding data source {dto.name} with config {json.dumps(dto.config)}")
- data_source = DataSourceContext.create_data_source(name=dto.name, config=dto.config)
+ data_source = await DataSourceContext.create_data_source(name=dto.name, config=dto.config)
  Posthog.added_data_source(uuid=request.headers.get('uuid'), name=dto.name)
  # in main.py we have a background task that runs every 5 minutes and indexes the data source
  # but here we want to index the data source immediately

app/clear_ack_queue.sh

@@ -0,0 +1 @@
+sqlite3 ~/.gerev/storage/tasks.sqlite3/data.db 'delete from ack_queue_task where _id in (select _id from ack_queue_task);'

app/clear_data_sources.sh

@@ -0,0 +1 @@
+sqlite3 ~/.gerev/storage/db.sqlite3 'delete from data_source where id in (select id from data_source);'

app/data_source/api/base_data_source.py

@@ -61,7 +61,7 @@ def get_config_fields() -> List[ConfigField]:
 
  @staticmethod
  @abstractmethod
- def validate_config(config: Dict) -> None:
+ async def validate_config(config: Dict) -> None:
  """
  Validates the config and raises an exception if it's invalid.
  """

app/data_source/api/context.py

@@ -6,9 +6,10 @@
 from data_source.api.base_data_source import BaseDataSource
 from data_source.api.dynamic_loader import DynamicLoader, ClassInfo
 from data_source.api.exception import KnownException
-from db_engine import Session
+from db_engine import Session, async_session
+from pydantic.error_wrappers import ValidationError
 from schemas import DataSourceType, DataSource
-
+from sqlalchemy import select
 
 logger = logging.getLogger(__name__)
 
@@ -48,22 +49,31 @@ def get_data_source_classes(cls) -> Dict[str, BaseDataSource]:
  return cls._data_source_classes
 
  @classmethod
- def create_data_source(cls, name: str, config: dict) -> BaseDataSource:
- with Session() as session:
- data_source_type = session.query(DataSourceType).filter_by(name=name).first()
+ async def create_data_source(cls, name: str, config: dict) -> BaseDataSource:
+ async with async_session() as session:
+ data_source_type = await session.execute(
+ select(DataSourceType).filter_by(name=name)
+ )
+ data_source_type = data_source_type.scalar_one_or_none()
  if data_source_type is None:
  raise KnownException(message=f"Data source type {name} does not exist")
 
  data_source_class = DynamicLoader.get_data_source_class(name)
  logger.info(f"validating config for data source {name}")
- data_source_class.validate_config(config)
+ await data_source_class.validate_config(config)
  config_str = json.dumps(config)
 
- data_source_row = DataSource(type_id=data_source_type.id, config=config_str, created_at=datetime.now())
+ data_source_row = DataSource(
+ type_id=data_source_type.id,
+ config=config_str,
+ created_at=datetime.now(),
+ )
  session.add(data_source_row)
- session.commit()
+ await session.commit()
 
- data_source = data_source_class(config=config, data_source_id=data_source_row.id)
+ data_source = data_source_class(
+ config=config, data_source_id=data_source_row.id
+ )
  cls._data_source_instances[data_source_row.id] = data_source
 
  return data_source
@@ -95,8 +105,12 @@ def _load_connected_sources_from_db(cls):
  for data_source in data_sources:
  data_source_cls = DynamicLoader.get_data_source_class(data_source.type.name)
  config = json.loads(data_source.config)
- data_source_instance = data_source_cls(config=config, data_source_id=data_source.id,
- last_index_time=data_source.last_indexed_at)
+ try:
+ data_source_instance = data_source_cls(config=config, data_source_id=data_source.id,
+ last_index_time=data_source.last_indexed_at)
+ except ValidationError as e:
+ logger.error(f"Error loading data source {data_source.id}: {e}")
+ return
  cls._data_source_instances[data_source.id] = data_source_instance
 
  cls._initialized = True

app/data_source/api/utils.py

@@ -4,12 +4,36 @@
 from functools import lru_cache
 from io import BytesIO
 from typing import Optional
+import time
+import threading
+from functools import wraps
 
 import requests
 
+
 logger = logging.getLogger(__name__)
 
 
+def rate_limit(*, allowed_per_second: int):
+ max_period = 1.0 / allowed_per_second
+ last_call = [time.perf_counter()]
+ lock = threading.Lock()
+
+ def decorate(func):
+ @wraps(func)
+ def limit(*args, **kwargs):
+ with lock:
+ elapsed = time.perf_counter() - last_call[0]
+ hold = max_period - elapsed
+ if hold > 0:
+ time.sleep(hold)
+ result = func(*args, **kwargs)
+ last_call[0] = time.perf_counter()
+ return result
+ return limit
+ return decorate
+
+
 def snake_case_to_pascal_case(snake_case_string: str):
  """Converts a snake case string to a PascalCase string"""
  components = snake_case_string.split('_')
@@ -55,3 +79,4 @@ def get_confluence_user_image(image_url: str, token: str) -> Optional[str]:
  return f"data:image/jpeg;base64,{base64.b64encode(image_bytes.getvalue()).decode()}"
  except:
  logger.warning(f"Failed to get confluence user image {image_url}")
+

app/data_source/sources/bookstack/bookstack.py

@@ -132,7 +132,7 @@ def list_books(book_stack: BookStack) -> List[Dict]:
  raise e
 
  @staticmethod
- def validate_config(config: Dict) -> None:
+ async def validate_config(config: Dict) -> None:
  try:
  parsed_config = BookStackConfig(**config)
  book_stack = BookStack(url=parsed_config.url, token_id=parsed_config.token_id,

app/data_source/sources/confluence/confluence.py

@@ -61,7 +61,7 @@ def list_all_spaces(confluence: Confluence) -> List[Location]:
  return spaces
 
  @staticmethod
- def validate_config(config: Dict) -> None:
+ async def validate_config(config: Dict) -> None:
  try:
  client = ConfluenceDataSource.confluence_client_from_config(config)
  ConfluenceDataSource.list_spaces(confluence=client)

app/data_source/sources/confluence/confluence_cloud.py

@@ -25,7 +25,7 @@ def get_config_fields() -> List[ConfigField]:
  ]
 
  @staticmethod
- def validate_config(config: Dict) -> None:
+ async def validate_config(config: Dict) -> None:
  try:
  client = ConfluenceCloudDataSource.confluence_client_from_config(config)
  ConfluenceCloudDataSource.list_spaces(confluence=client)

app/data_source/sources/google_drive/google_drive.py

@@ -43,7 +43,7 @@ def get_config_fields() -> List[ConfigField]:
  ]
 
  @staticmethod
- def validate_config(config: Dict) -> None:
+ async def validate_config(config: Dict) -> None:
  try:
  scopes = ['https://www.googleapis.com/auth/drive.readonly']
  parsed_config = GoogleDriveConfig(**config)

app/data_source/sources/mattermost/mattermost.py

@@ -54,7 +54,7 @@ def get_config_fields() -> List[ConfigField]:
  ]
 
  @staticmethod
- def validate_config(config: Dict) -> None:
+ async def validate_config(config: Dict) -> None:
  try:
  parsed_config = MattermostConfig(**config)
  maattermost = Driver(options=asdict(parsed_config))

app/data_source/sources/rocketchat/rocketchat.py

@@ -54,7 +54,7 @@ def get_display_name(cls) -> str:
  return "Rocket.Chat"
 
  @staticmethod
- def validate_config(config: Dict) -> None:
+ async def validate_config(config: Dict) -> None:
  rocket_chat_config = RocketchatConfig(**config)
  should_verify_ssl = os.environ.get('ROCKETCHAT_VERIFY_SSL') is not None
  rocket_chat = RocketChat(user_id=rocket_chat_config.token_id, auth_token=rocket_chat_config.token_secret,

app/data_source/sources/slack/slack.py

@@ -42,7 +42,7 @@ def get_config_fields() -> List[ConfigField]:
  ]
 
  @staticmethod
- def validate_config(config: Dict) -> None:
+ async def validate_config(config: Dict) -> None:
  slack_config = SlackConfig(**config)
  slack = WebClient(token=slack_config.token)
  slack.auth_test()

app/data_source/sources/stackoverflow/stackoverflow.py

@@ -0,0 +1,136 @@
+import logging
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Dict, List, Optional
+import requests
+
+from data_source.api.base_data_source import BaseDataSource, ConfigField, HTMLInputType, BaseDataSourceConfig
+from data_source.api.basic_document import DocumentType, BasicDocument
+from queues.index_queue import IndexQueue
+
+from data_source.api.utils import rate_limit
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class StackOverflowPost:
+ link: str
+ score: int
+ last_activity_date: int
+ creation_date: int
+ post_id: Optional[int] = None
+ post_type: Optional[str] = None
+ body_markdown: Optional[str] = None
+ owner_account_id: Optional[int] = None
+ owner_reputation: Optional[int] = None
+ owner_user_id: Optional[int] = None
+ owner_user_type: Optional[str] = None
+ owner_profile_image: Optional[str] = None
+ owner_display_name: Optional[str] = None
+ owner_link: Optional[str] = None
+ title: Optional[str] = None
+ last_edit_date: Optional[str] = None
+ tags: Optional[List[str]] = None
+ view_count: Optional[int] = None
+ article_id: Optional[int] = None
+ article_type: Optional[str] = None
+
+class StackOverflowConfig(BaseDataSourceConfig):
+ api_key: str
+ team_name: str
+
+
+@rate_limit(allowed_per_second=15)
+def rate_limited_get(url, headers):
+ '''
+ https://api.stackoverflowteams.com/docs/throttle
+ https://api.stackexchange.com/docs/throttle
+ Every application is subject to an IP based concurrent request throttle.
+ If a single IP is making more than 30 requests a second, new requests will be dropped.
+ The exact ban period is subject to change, but will be on the order of 30 seconds to a few minutes typically.
+ Note that exactly what response an application gets (in terms of HTTP code, text, and so on)
+ is undefined when subject to this ban; we consider > 30 request/sec per IP to be very abusive and thus cut the requests off very harshly.
+ '''
+ resp = requests.get(url, headers=headers)
+ if resp.status_code == 429:
+ logger.warning('Rate limited, sleeping for 5 minutes')
+ time.sleep(300)
+ return rate_limited_get(url, headers)
+ return resp
+
+
+class StackOverflowDataSource(BaseDataSource):
+
+ @staticmethod
+ def get_config_fields() -> List[ConfigField]:
+ return [
+ ConfigField(label="PAT API Key", name="api_key", type=HTMLInputType.TEXT),
+ ConfigField(label="Team Name", name="team_name", type=HTMLInputType.TEXT),
+ ]
+
+ @staticmethod
+ async def validate_config(config: Dict) -> None:
+ so_config = StackOverflowConfig(**config)
+ url = f'https://api.stackoverflowteams.com/2.3/questions?&team={so_config.team_name}'
+ response = rate_limited_get(url, headers={'X-API-Access-Token': so_config.api_key})
+ response.raise_for_status()
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ so_config = StackOverflowConfig(**self._raw_config)
+ self._api_key = so_config.api_key
+ self._team_name = so_config.team_name
+
+ def _fetch_posts(self, *, api_key: str, team_name: str, page: int, doc_type: str) -> None:
+ team_fragment = f'&team={team_name}'
+ # this is a filter for "body markdown" inclusion, all filters are unique and static
+ # i am not entirely sure if this is per account, or usable by everyone
+ filter_fragment = '&filter=!nOedRLbqzB'
+ page_fragment = f'&page={page}'
+ # it looked like the timestamp was 10 digits, lets only look at stuff that is newer than the last index time
+ from_date_fragment = f'&fromdate={self._last_index_time.timestamp():.10n}'
+ url = f'https://api.stackoverflowteams.com/2.3/{doc_type}?{team_fragment}{filter_fragment}{page_fragment}{from_date_fragment}'
+ response = rate_limited_get(url, headers={'X-API-Access-Token': api_key})
+ response.raise_for_status()
+ response = response.json()
+ has_more = response['has_more']
+ items = response['items']
+ logger.info(f'Fetched {len(items)} {doc_type} from Stack Overflow')
+ for item_dict in items:
+ owner_fields = {}
+ if 'owner' in item_dict:
+ owner_fields = {f"owner_{k}": v for k, v in item_dict.pop('owner').items()}
+ if 'title' not in item_dict:
+ item_dict['title'] = item_dict['link']
+ post = StackOverflowPost(**item_dict, **owner_fields)
+ last_modified = datetime.fromtimestamp(post.last_edit_date or post.last_activity_date)
+ if last_modified < self._last_index_time:
+ return
+ logger.info(f'Feeding {doc_type} {post.title}')
+ post_document = BasicDocument(title=post.title, content=post.body_markdown, author=post.owner_display_name,
+ timestamp=datetime.fromtimestamp(post.creation_date), id=post.post_id,
+ data_source_id=self._data_source_id, location=post.link,
+ url=post.link, author_image_url=post.owner_profile_image,
+ type=DocumentType.MESSAGE)
+ IndexQueue.get_instance().put_single(doc=post_document)
+ if has_more:
+ # paginate onto the queue
+ self.add_task_to_queue(self._fetch_posts, api_key=self._api_key, team_name=self._team_name, page=page + 1, doc_type=doc_type)
+
+ def _feed_new_documents(self) -> None:
+ self.add_task_to_queue(self._fetch_posts, api_key=self._api_key, team_name=self._team_name, page=1, doc_type='posts')
+ # TODO: figure out how to get articles
+ # self.add_task_to_queue(self._fetch_posts, api_key=self._api_key, team_name=self._team_name, page=1, doc_type='articles')
+
+
+# def test():
+# import os
+# config = {"api_key": os.environ['SO_API_KEY'], "team_name": os.environ['SO_TEAM_NAME']}
+# so = StackOverflowDataSource(config=config, data_source_id=1)
+# so._feed_new_documents()
+#
+#
+# if __name__ == '__main__':
+# test()

app/db_engine.py

@@ -5,11 +5,19 @@
 
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
 # import base document and then register all classes
 from schemas.base import Base
 
 from paths import SQLITE_DB_PATH
 
-engine = create_engine(f'sqlite:///{SQLITE_DB_PATH}')
+db_url = f'sqlite:///{SQLITE_DB_PATH}'
+print('DB engine path:', db_url)
+engine = create_engine(db_url)
 Base.metadata.create_all(engine)
 Session = sessionmaker(bind=engine)
+
+async_db_url = db_url.replace('sqlite', 'sqlite+aiosqlite', 1)
+print('ASYNC DB engine path:', async_db_url)
+async_engine = create_async_engine(async_db_url)
+async_session = sessionmaker(async_engine, expire_on_commit=False, class_=AsyncSession)