Build your own LLM RAG API without langchain using OpenAI, qdrant and PDF stack

brew install pyenv

curl https://pyenv.run | bash

# For bash:
echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.bashrc
echo 'command -v pyenv >/dev/null || export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.bashrc
echo 'eval "$(pyenv init -)"' >> ~/.bashrc

# For Zsh:
echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.zshrc
echo 'command -v pyenv >/dev/null || export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.zshrc
echo 'eval "$(pyenv init -)"' >> ~/.zshrc

pyenv install 3.11

mkdir bRAG
cd bRAG

pyenv virtualenv 3.11 bRAG-3-11  # this creates a 'virtualenv' based on python 3.11 named bRAG-3-11
pyenv local bRAG-3-11

python -m pip install pre-commit pip-tools

[project]
name = "brag"
version = "0.0.1"
requires-python = ">=3.11"
dependencies = ["fastapi", "openai", "pydantic-settings", "uvicorn", "gunicorn", "qdrant-client[fastembed]", "sentry-sdk", "pugsql", "psycopg2-binary"]


[project.optional-dependencies]
dev = ["ruff", "black"]

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

 pip-compile --generate-hashes -o requirements/requirements.txt pyproject.toml  # for production
 pip-compile --extra dev -o requirements/requirements-dev.txt pyproject.toml  # local usage

pip-sync requirements/requirements.txt requirements/requirements-dev.txt

 |-- app/  # Main codebase directory
 |---- chat/
 |------ __init__.py
 |------ api.py
 |------ constants.py
 |------ message.py
 |------ models.py
 |------ streams.py
 |---- core/
 |------ __init__.py
 |------ api.py
 |------ logs.py
 |------ middlewares.py
 |------ models.py
 |---- __init__.py
 |---- db.py
 |---- main.py


 |-- db/  # Database/migration/pugsql related code
 |---- migrations/
 |---- queries/
 |---- schema.sql
 |-- settings/  # Settings files directory
 |---- base.py  # base config for our app
 |---- gunicorn.conf.py  # config for gunicorn
 |-- tests/  # Main tests directory
 |-- requirements/
 |-- .dockerignore
 |-- .env.example
 |-- .gitignore
 |-- pre-commit-config.yaml  # for linting in during development
 |-- docker-compose.yml
 |-- Dockerfile
 |-- Makefile  # useful shortcuts
 |-- README.md

 bRAG
 |-- app/
 |---- core/
 |-- main.py

# app/core/logs.py
import logging

# we are setting the logging level to log everything at level info and above
logging.basicConfig(level=logging.INFO)

# we are creating one common logger for our application that we will later use
# everywhere we want to log stuff
# this provides common configuration and, if we want it, standard/format
logger = logging.getLogger(__name__)
# we will push the logs to the console/stdout
ConsoleOutputHandler = logging.StreamHandler()
logger.addHandler(ConsoleOutputHandler)


# main.py - at the top level of the project
from fastapi import FastAPI

from app.core.logs import logger

# in here we just initialise the fastapi app, without defining
# any endpoints
app = FastAPI()
logger.info("App is ready!")

# __init__.py - at the top level of the project
# read-only tomllib is now part of python core, so we do not need to install it
import tomllib

# we get the version from pyproject.toml and put it into a variable
with open("pyproject.toml", "rb") as f:
    data = tomllib.load(f)
    version = data["project"]["version"]


# main.py - at the top level of the project
from fastapi import FastAPI

from app import version
from app.core.logs import logger

app = FastAPI(version=version)

logger.info("App is ready!")

# core/api.py
from fastapi import APIRouter

from starlette import status
from starlette.requests import Request

# we define a router to hold all our core endpoints together
router = APIRouter(tags=["Core Endpoints"])  

@router.get("/healthcheck", status_code=status.HTTP_200_OK)
async def healthcheck(request: Request) -> dict:
    return {"version": request.app.version}

# main.py
from fastapi import FastAPI

from app import version
from app.core.api import router as core_router
from app.core.logs import logger

app = FastAPI(version=version)
app.include_router(core_router)
logger.info("App is ready!")

python -m app.main

python -m uvicorn --reload app.main:app

# core/constants.py
from enum import StrEnum

# on local we just need 1 worker, no need to spawn multiple threads
DEFAULT_NUMBER_OF_WORKERS_ON_LOCAL = 1  


# we could have any number of environments here, this is useful when
# defining settings/behavior per environment - eg on production
# you probably do not want debug enabled and so on.
class Environments(StrEnum):
    LOCAL = "local"
    PRODUCTION = "production"

# settings/base.py
from pydantic import Field
from pydantic_settings import BaseSettings

from app.core.constants import Environments


class Settings(BaseSettings):
    APP_NAME: str = "bRAG"
    ENVIRONMENT: str = Field(env="ENVIRONMENT", default=Environments.LOCAL.value)
    OPENAI_API_KEY: str = Field(env="OPENAI_API_KEY")

    @property
    def is_local(self):
        return self.ENVIRONMENT == Environments.LOCAL.value

    # this will make it so that pydantic will look for our env variables
    # in a file called .env and automatically load them as such
    class Config:
        env_file = ".env"

# settings/__init__.py
from settings.base import Settings

# this enables us to easily import settings later by going 
# from settings import settings
# something like django settings
settings = Settings()

# .env.example 
OPENAI_API_KEY=sk-xxxxxx

# chat/constants.py
from enum import StrEnum


class FailureReasonsEnum(StrEnum):
    OPENAI_ERROR = "OpenAI call failed"
    STREAM_TIMEOUT = "Stream timed out"  # for later
    FAILED_PROCESSING = "Post processing failed"  # for later


class ChatRolesEnum(StrEnum):
    USER = "user"
    SYSTEM = "system"
    ASSISTANT = "assistant"


class ModelsEnum(StrEnum):
    GPT4 = "gpt-4-0613"


# chat/models.py
from pydantic import BaseModel, Field

from app.chat.constants import ChatRolesEnum, ModelsEnum
from app.core.models import TimestampAbstractModel


class BaseMessage(BaseModel):
    """ Base pydantic model that we use to interact with the API."""
    model: ModelsEnum = Field(default=ModelsEnum.GPT4.value)
    message: str


class Message(TimestampAbstractModel, BaseMessage):
    role: ChatRolesEnum



class Message(BaseMessage):
    role: ChatRolesEnum

# core/models.py
from datetime import datetime

from pydantic import BaseModel, Field


# utility base class to add creation time to all the models
# usually we would also have updated_at field to denote
# when the object was updated, but in our case I doubt we will do
# any updates
class TimestampAbstractModel(BaseModel):
    created_at: datetime = Field(default_factory=datetime.utcnow)

# chat/api.py

from fastapi import APIRouter

import openai

from app.chat.exceptions import OpenAIException
from app.chat.models import BaseMessage, Message
from app.chat.services import OpenAIService


router = APIRouter(tags=["Core Endpoints"])


@router.post("/v1/completion")
async def completion_create(input_message: BaseMessage) -> Message:
    try:
        return await OpenAIService.chat_completion(input_message=input_message)
    except openai.OpenAIError:
        raise OpenAIException
# main.py
# do not forget to include the new router:
(...)
from app.chat.api import router as chat_router
from app.core.api import router as core_router
app = FastAPI(version=version)
(...)
app.include_router(core_router)
app.include_router(chat_router)
(...)

# chat/services.py
import openai
from openai import ChatCompletion

from app.chat.constants import ChatRolesEnum
from app.chat.models import BaseMessage, Message
from app.core.logs import logger
from settings import settings


class OpenAIService:
    @classmethod
    async def chat_completion(cls, input_message: BaseMessage) -> Message:
        logger.info(f"Received the following completion: {input_message}")
        completion: openai.ChatCompletion = await openai.ChatCompletion.acreate(
            model=input_message.model,
            api_key=settings.OPENAI_API_KEY,
            messages=[{"role": ChatRolesEnum.ASSISTANT.value, "content": input_message.message}],
        )
        logger.info(f"Got the following response: {completion}")
        return Message(
            model=input_message.model,
            message=cls.extract_response_from_completion(completion),
            role=input_message.role,
        )

    @staticmethod
    def extract_response_from_completion(chat_completion: ChatCompletion) -> str:
        return chat_completion.choices[0]["message"]["content"]

# chats/exceptions.py
# here we define some exceptions that will come handy in the future
from fastapi import HTTPException
from starlette import status

from app.chat.constants import FailureReasonsEnum, NO_DOCUMENTS_FOUND


class OpenAIException(HTTPException):
    def __init__(self):
        super().__init__(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=FailureReasonsEnum.OPENAI_ERROR.value)


class OpenAIFailedProcessingException(HTTPException):
    def __init__(self):
        super().__init__(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=FailureReasonsEnum.FAILED_PROCESSING.value)


class OpenAIStreamTimeoutException(HTTPException):
    def __init__(self):
        super().__init__(status_code=status.HTTP_504_GATEWAY_TIMEOUT, detail=FailureReasonsEnum.STREAM_TIMEOUT.value)


class RetrievalNoDocumentsFoundException(HTTPException):
    def __init__(self):
        super().__init__(status_code=status.HTTP_404_NOT_FOUND, detail=)


# chats/streaming.py
import asyncio
import json

import async_timeout

from app.chat.constants import ChatRolesEnum
from app.chat.exceptions import OpenAIStreamTimeoutException, OpenAIFailedProcessingException
from app.chat.models import Chunk, Message
from app.core.logs import logger
from settings import settings


async def stream_generator(subscription):
    """In the future if we are paranoid about performance we can pre-allocate a buffer and use it instead of f-string
    should save a cycle or two in the processor or be a fun exercise but totally not worth the effort or complexity
    currently (in the future too).
    """
    async with async_timeout.timeout(settings.GENERATION_TIMEOUT_SEC):
        try:
            complete_response: str = ""
            async for chunk in subscription:
                # f-string is faster than concatenation so we use it here
                complete_response = f"{complete_response}{Chunk.get_chunk_delta_content(chunk=chunk)}"
                yield format_to_event_stream(post_processing(chunk))
            message: Message = Message(model=chunk.model, message=complete_response, role=ChatRolesEnum.ASSISTANT.value)
            logger.info(f"Complete Streamed Message: {message}")
        except asyncio.TimeoutError:
            raise OpenAIStreamTimeoutException


def format_to_event_stream(data: str) -> str:
    """ We format the event to a format more in line with the standard SSE format. """
    return f"event: message\ndata: {data}\n\n"


def post_processing(chunk) -> str:
    try:
        logger.info(f"Chunk: {chunk}")
        formatted_chunk = Chunk.from_chunk(chunk=chunk)
        logger.info(f"Formatted Chunk: {formatted_chunk}")
        return json.dumps(formatted_chunk.model_dump())
    except Exception:
        raise OpenAIFailedProcessingException

# chats/services.py
class OpenAIService:
    @classmethod
    async def chat_completion_without_streaming(cls, input_message: BaseMessage) -> Message:
        completion: openai.ChatCompletion = await openai.ChatCompletion.acreate(
            model=input_message.model,
            api_key=settings.OPENAI_API_KEY,
            messages=[{"role": ChatRolesEnum.USER.value, "content": input_message.message}],
        )
        logger.info(f"Got the following response: {completion}")
        return Message(
            model=input_message.model,
            message=cls.extract_response_from_completion(completion),
            role=ChatRolesEnum.ASSISTANT.value
        )

    @staticmethod
    async def chat_completion_with_streaming(input_message: BaseMessage) -> StreamingResponse:
        subscription: openai.ChatCompletion = await openai.ChatCompletion.acreate(
            model=input_message.model,
            api_key=settings.OPENAI_API_KEY,
            messages=[{"role": ChatRolesEnum.USER.value, "content": input_message.message}],
            stream=True,
        )
        return StreamingResponse(stream_generator(subscription), media_type="text/event-stream")

    @staticmethod
    def extract_response_from_completion(chat_completion: ChatCompletion) -> str:
        return chat_completion.choices[0]["message"]["content"]

from fastapi import APIRouter

import openai
from starlette.responses import StreamingResponse

from app.chat.exceptions import OpenAIException
from app.chat.models import Message, BaseMessage
from app.chat.services import OpenAIService


router = APIRouter(tags=["Chat Endpoints"])


@router.post("/v1/completion")
async def completion_create(input_message: BaseMessage) -> Message:
    try:
        return await OpenAIService.chat_completion_without_streaming(input_message=input_message)
    except openai.OpenAIError:
        raise OpenAIException


@router.post("/v1/completion-stream")
async def completion_stream(input_message: BaseMessage) -> StreamingResponse:
    """ Streaming response won't return json but rather a properly formatted string for SSE."""
    try:
        return await OpenAIService.chat_completion_with_streaming(input_message=input_message)
    except openai.OpenAIError:
        raise OpenAIException

docs = [
  "Qdrant has Langchain integrations", 
  "Qdrant also has Llama Index integrations"
]
client.add(
  collection_name="demo_collection",
  documents=docs,
)

search_result = client.query(
  collection_name="demo_collection",
  query_text="This is a query document"
)

docker pull qdrant/qdrant
docker run -p 6333:6333 qdrant/qdrant

from qdrant_client import QdrantClient

from settings import settings


def test_populate_vector_db(client: QdrantClient):
    # Prepare your documents, metadata, and IDs
    docs = (
        "LoremIpsum sit dolorem",
        "Completely random phrase",
        "Another random phrase",
        "pdf-BRAG is awesome.",
    )
    metadata = (
        {"source": "Olaf"},
        {"source": "grski"},
        {"source": "Olaf"},
        {"source": "grski"},
    )
    ids = [42, 2, 3, 5]

    # Use the new add method
    client.add(
        collection_name=settings.QDRANT_COLLECTION_NAME,
        documents=docs,
        metadata=metadata,
        ids=ids,
    )

# settings/base.py
class Settings(BaseSettings):
    APP_NAME: str = "bRAG"
    ENVIRONMENT: str = Field(env="ENVIRONMENT", default=Environments.LOCAL.value)

    DATABASE_URL: str = Field(env="DATABASE_URL")
    OPENAI_API_KEY: str = Field(env="OPENAI_API_KEY", default="None")
    GENERATION_TIMEOUT_SEC: int = Field(env="GENERATION_TIMEOUT_SEC", default=120)

    QDRANT_HOST: str = Field(env="QDRANT_HOST", default="localhost")
    QDRANT_PORT: int = Field(env="QDRANT_PORT", default=6333)
    QDRANT_COLLECTION_NAME: str = Field(env="QDRANT_COLLECTION_NAME", default="demo")

    @property
    def is_local(self):
        return self.ENVIRONMENT == Environments.LOCAL.value

    class Config:
        env_file = ".env"


# chats/retrieval.py
from qdrant_client import QdrantClient

from app.chat.exceptions import RetrievalNoDocumentsFoundException
from app.chat.models import BaseMessage
from settings import settings

client = QdrantClient(settings.QDRANT_HOST, port=settings.QDRANT_PORT)

# ONLY FOR TEST PURPOSE (you may call this func by hand or let it be)
test_populate_vector_db(client=client)


def process_retrieval(message: BaseMessage) -> BaseMessage:
    """ Search for a given query using vector similarity search. If no documents are found we raise an exception.
    If we do find documents we take the top 3 and put them into the context."""
    search_result = search(query=message.message)
    resulting_query: str = (
        f"Answer based only on the context, nothing else. \n"
        f"QUERY:\n{message.message}\n"
        f"CONTEXT:\n{search_result}"
    )
    logger.info(f"Resulting Query: {resulting_query}")
    return BaseMessage(message=resulting_query, model=message.model)


def search(query: str) -> str:
    """ This takes our query string, transforms the string into vectors and then searches for the most similar
    documents, returning the top 3. If no documents are found we raise an exception - as the user is asking
    about something not in the context of our documents."""
    search_result = client.query(collection_name=settings.QDRANT_COLLECTION_NAME, limit=3, query_text=query)
    if not search_result:
        raise RetrievalNoDocumentsFoundException
    return "\n".join(result.payload["page_content"] for result in search_result)

# chat/constants.py
NO_DOCUMENTS_FOUND: str = "No documents found in context. Please try again with a different query."

# chat/services.py 
from app.chat.constants import ChatRolesEnum, NO_DOCUMENTS_FOUND
(...)

OpenAIService (...)
    @classmethod
    async def qa_without_stream(cls, input_message: BaseMessage) -> Message:
        try:
            augmented_message: BaseMessage = process_retrieval(message=input_message)
            return await cls.chat_completion_without_streaming(input_message=augmented_message)
        except RetrievalNoDocumentsFoundException:
            return Message(model=input_message.model, message=NO_DOCUMENTS_FOUND, role=ChatRolesEnum.ASSISTANT.value)

@router.post("/v1/qa-create")
async def qa_create(input_message: BaseMessage) -> Message:
    try:
        return await OpenAIService.qa_without_stream(input_message=input_message)
    except openai.OpenAIError:
        raise OpenAIException

or on mac go with:

```bash
brew install dbmate

dbmate new create_models # dbmate new <desciprtion_of_the_migrations>

-- migrate:up
<migration code goes here>

-- migrate:down
<rollback code goes here>

-- migrate:up

CREATE TABLE message (
    id SERIAL PRIMARY KEY,
    model VARCHAR(255),
    role VARCHAR(255),
    message TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW(),
);

-- migrate:down

DROP TABLE IF EXISTS message;

  database:
    restart: always
    image: postgres:15-alpine
    volumes:
      - pg-data:/var/lib/postgresql/data
    healthcheck:
      test: [ "CMD-SHELL", "pg_isready -U application" ]
      interval: 5s
      timeout: 5s
      retries: 5
    environment:
      POSTGRES_USER: application
      POSTGRES_PASSWORD: secret_pass
      POSTGRES_DB: application
    ports:
      - "5432:5432"

dbmate up

  dbmate:
    image: amacneil/dbmate:latest
    command:
      - "up"
    depends_on:
      database:
        condition: service_healthy

  api:
    build:
      context: .
    depends_on:
      database:
        condition: service_healthy
    restart: always
    deploy:
      replicas: 1
    ports:
      - "8000:8000"
    volumes:
      - .:/code/
    command: ["bash", "-c","dbmate up && gunicorn -c settings/gunicorn.conf.py app.main:app"]
    env_file:
      - .env

DATABASE_URL="postgresql://application:secret_pass@database:5432/application?sslmode=disable"
# or
protocol://username:password@host:port/database_name?options

#make note that instead of `database` you might need to tpye in `localhost` depending where you are running - on localhost or inside docker container/network

SELECT * FROM pg_tables WHERE schemaname = 'public';

public,schema_migrations,application,,true,false,false,false
public,message,application,,true,false,false,false

# db/queries/messages/select_all.sql
-- :name select_all :many
SELECT * FROM message ORDER BY created_at DESC

# db/queries/messages/insert.sql
-- :name insert :insert
INSERT INTO message (model, role, message)
VALUES (:model, :role, :message);

# app/db.py
import pugsql

messages_queries = pugsql.module("db/queries/messages")
messages_connection = messages_queries.connect(settings.DATABASE_URL)

from fastapi import APIRouter

import openai
from starlette.responses import StreamingResponse

from app.chats.exceptions import OpenAIException
from app.chats.models import Message, BaseMessage
from app.chats.services import OpenAIService
from app.db import messages_queries

router = APIRouter(tags=["Chat Endpoints"])


@router.get("/v1/messages")
async def get_messages() -> list[Message]:
    # a bit messy as we might want to move this to a service
    return [Message(**message) for message in messages_queries.select_all()]

class OpenAIService:
    @classmethod
    async def chat_completion_without_streaming(cls, input_message: BaseMessage) -> Message:
        completion: openai.ChatCompletion = await openai.ChatCompletion.acreate(
            model=input_message.model,
            api_key=settings.OPENAI_API_KEY,
            messages=[{"role": ChatRolesEnum.USER.value, "content": input_message.message}],
        )
        logger.info(f"Got the following response: {completion}")
        message = Message(
            model=input_message.model,
            message=cls.extract_response_from_completion(completion),
            role=ChatRolesEnum.ASSISTANT.value
        )
        messages_queries.insert(model=message.model, message=message.message, role=message.role)
        return message
# or for saving the complete response after streaming:

    async with async_timeout.timeout(settings.GENERATION_TIMEOUT_SEC):
        try:
            complete_response: str = ""
            async for chunk in subscription:
                # f-string is faster than concatenation so we use it here
                complete_response = f"{complete_response}{Chunk.get_chunk_delta_content(chunk=chunk)}"
                yield format_to_event_stream(post_processing(chunk))
            message: Message = Message(model=chunk.model, message=complete_response, role=ChatRolesEnum.ASSISTANT.value)
            messages_queries.insert(model=message.model, message=message.message, role=message.role)
            logger.info(f"Complete Streamed Message: {message}")
        except asyncio.TimeoutError:
            raise OpenAIStreamTimeoutException

# Dockerfile
FROM python:3.11.6 as server
RUN curl -fsSL -o /usr/local/bin/dbmate https://github.com/amacneil/dbmate/releases/latest/download/dbmate-linux-amd64 && \
    chmod +x /usr/local/bin/dbmate
RUN pip install --no-cache-dir --upgrade pip
WORKDIR /project
COPY /requirements/requirements.txt /project/requirements.txt
RUN pip install --no-cache-dir --upgrade -r /project/requirements.txt
COPY . /project
CMD ["bash", "-c", "dbmate up & gunicorn -c settings/gunicorn.conf.py app.main:app"]

# pre-commit-config.yaml
repos:
-   repo: https://github.com/psf/black
    rev: 23.3.0
    hooks:
    - id: black
      language_version: python3.11

- repo: https://github.com/charliermarsh/ruff-pre-commit
  rev: 'v0.0.274'
  hooks:
    - id: ruff
      args: ["--fix"]

PATH  := $(PATH)
SHELL := /bin/bash

black:
    black .

ruff:
    ruff check . --fix

lint:
    make black
    make ruff

pre-commit:
    pre-commit run --all-files

test-without-clean:
    set -o pipefail; \
    pytest \
      --color=yes \
      --junitxml=pytest.xml \
      --cov-report=term-missing:skip-covered \
      --cov=app \
      tests \
      | tee pytest-coverage.txt

test:
    make test-without-clean
    make clean

upgrade:
    make pip-compile-upgrade
    make pip-compile-dev-upgrade


compile:
    make pip-compile
    make pip-compile-dev

sync:
    pip-sync requirements/requirements.txt
sync-dev:
    pip-sync requirements/requirements.txt requirements/requirements-dev.txt

pip-compile:
    pip-compile -o requirements/requirements.txt pyproject.toml
pip-compile-dev:
    pip-compile --extra dev -o requirements/requirements-dev.txt pyproject.toml

pip-compile-upgrade:
    pip-compile --strip-extras --upgrade -o requirements/requirements.txt pyproject.toml
pip-compile-dev-upgrade:
    pip-compile --extra dev --upgrade -o requirements/requirement-dev.txt pyproject.toml

run:
    python -m gunicorn -c settings/gunicorn.conf.py app.main:app

run-dev:
    python -m uvicorn --reload app.main:app

clean:
    rm -f pytest.xml
    rm -f pytest-coverage.txt

[tool.black]
line-length = 120
target-version = ['py311']
include = '\.pyi?$'
exclude = '''
/(
    \.eggs
  | \.git
  | \.hg
  | \.mypy_cache
  | \.tox
  | \.venv
  | _build
  | buck-out
  | build
  | dist
  | migrations
)/
'''

[tool.ruff]
line-length = 120
select = ["E", "F", "I"]
fixable = ["A", "B", "C", "D", "E", "F", "I"]
ignore = ["E712"]
exclude = [
    ".bzr",
    ".direnv",
    ".eggs",
    ".git",
    ".hg",
    ".mypy_cache",
    ".nox",
    ".pants.d",
    ".pytype",
    ".ruff_cache",
    ".svn",
    ".tox",
    ".venv",
    "__pypackages__",
    "_build",
    "buck-out",
    "build",
    "dist",
    "node_modules",
    "venv",
    "*migrations*",
]

[tool.ruff.isort]
section-order = ["fastapi", "future", "standard-library", "third-party",  "first-party", "local-folder"]

[tool.ruff.isort.sections]
fastapi = ["fastapi"]

# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class

# C extensions
*.so
.idea
.env

# Distribution / packaging
.Python
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST


# Installer logs
pip-log.txt
pip-delete-this-directory.txt

# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
pytest.xml
pytest-coverage.txt

# Translations
*.mo
*.pot

# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal

profile_default/
ipython_config.py
.python-version


# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/

# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/

# Spyder project settings
.spyderproject
.spyproject

# Rope project settings
.ropeproject

# mkdocs documentation
/site
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# Ruff
.ruff_cache/