feat: use only sqlalchemy for orm (#5)

Makefile

@@ -17,6 +17,19 @@ rm-db:
 	docker kill $$(docker ps -aqf name=postgres-fastqueue)
 	docker container rm $$(docker ps -aqf name=postgres-fastqueue)
 
+run-test-db:
+	docker run --name postgres-fastqueue-test \
+		--restart unless-stopped \
+		-e POSTGRES_USER=fastqueue \
+		-e POSTGRES_PASSWORD=fastqueue \
+		-e POSTGRES_DB=fastqueue-test \
+		-p 5432:5432 \
+		-d postgres:14-alpine
+
+rm-test-db:
+	docker kill $$(docker ps -aqf name=postgres-fastqueue-test)
+	docker container rm $$(docker ps -aqf name=postgres-fastqueue-test)
+
 build-image:
 	docker build --rm -t fastqueue .
 
@@ -29,4 +42,4 @@ run-db-migrate:
 create-auto-migration:
 	poetry run alembic revision --autogenerate -m "Auto generated"
 
-.PHONY: test lint run-db rm-db build-image run-server run-db-migrate create-auto-migration
+.PHONY: test lint run-db rm-db run-test-db rm-test-db build-image run-server run-db-migrate create-auto-migration

alembic/env.py

@@ -4,7 +4,7 @@
 
 from alembic import context
 from fastqueue.config import settings
-from fastqueue.models import metadata
+from fastqueue.models import Base
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
@@ -19,7 +19,7 @@
 # for 'autogenerate' support
 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
-target_metadata = metadata
+target_metadata = Base.metadata
 
 # other values from the config, defined by the needs of env.py,
 # can be acquired:

alembic/versions/001_initial.py

@@ -1,8 +1,8 @@
 """Auto generated
 
-Revision ID: 1f5fd70b6d4d
+Revision ID: 4485cf5c36b9
 Revises:
-Create Date: 2022-08-29 15:17:54.064031
+Create Date: 2022-08-29 18:44:21.413531
 
 """
 import sqlalchemy as sa
@@ -11,7 +11,7 @@
 from alembic import op
 
 # revision identifiers, used by Alembic.
-revision = "1f5fd70b6d4d"
+revision = "4485cf5c36b9"
 down_revision = None
 branch_labels = None
 depends_on = None

env.sample

@@ -8,7 +8,4 @@ fastqueue_server_port='8000'
 fastqueue_server_reload='true'
 fastqueue_server_num_workers='1'
 
-fastqueue_postgresql_host='localhost'
-fastqueue_postgresql_dbname='fastqueue'
-fastqueue_postgresql_user='fastqueue'
-fastqueue_postgresql_password='fastqueue'
+fastqueue_database_url='postgresql+psycopg2://fastqueue:fastqueue@localhost:5432/fastqueue-test'

fastqueue/config.py

@@ -20,30 +20,12 @@ class Settings(BaseSettings):
     server_num_workers: int = 1
 
     # postgresql settings
-    postgresql_host: str
-    postgresql_dbname: str
-    postgresql_user: str
-    postgresql_password: str
+    database_url: str
 
     class Config:
         env_file = ".env"
         env_file_encoding = "utf-8"
         env_prefix = "fastqueue_"
 
-    def _create_database_url(self, prefix: str) -> str:
-        user = self.postgresql_user
-        password = self.postgresql_password
-        host = self.postgresql_host
-        dbname = self.postgresql_dbname
-        return f"{prefix}://{user}:{password}@{host}/{dbname}"
-
-    @property
-    def database_url(self) -> str:
-        return self._create_database_url("postgresql+psycopg2")
-
-    @property
-    def async_database_url(self) -> str:
-        return self._create_database_url("postgresql+asyncpg")
-
 
 settings = Settings()

fastqueue/database.py

@@ -1,15 +1,18 @@
 from pathlib import Path
 
-from databases import Database
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
 
 from alembic import command
 from alembic.config import Config
 from fastqueue.config import settings
 from fastqueue.logger import get_logger
 
 logger = get_logger(__name__)
-force_rollback = True if settings.testing else False
-database = Database(settings.async_database_url, force_rollback=force_rollback)
+engine = create_engine(settings.database_url)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base = declarative_base()
 
 
 def run_migrations() -> None:

fastqueue/models.py

@@ -1,54 +1,52 @@
 import sqlalchemy
-from sqlalchemy.dialects.postgresql import JSONB, UUID
-
-metadata = sqlalchemy.MetaData()
-
-topics = sqlalchemy.Table(
-    "topics",
-    metadata,
-    sqlalchemy.Column("id", sqlalchemy.String(length=128), primary_key=True, nullable=False),
-    sqlalchemy.Column("created_at", sqlalchemy.DateTime, nullable=False),
-)
-
-queues = sqlalchemy.Table(
-    "queues",
-    metadata,
-    sqlalchemy.Column("id", sqlalchemy.String(length=128), primary_key=True, nullable=False),
-    sqlalchemy.Column(
-        "topic_id",
-        sqlalchemy.String(length=128),
-        sqlalchemy.ForeignKey("topics.id"),
-        index=True,
-        nullable=False,
-    ),
-    sqlalchemy.Column("ack_deadline_seconds", sqlalchemy.Integer, nullable=False),
-    sqlalchemy.Column("message_retention_seconds", sqlalchemy.Integer, nullable=False),
-    sqlalchemy.Column("message_filters", JSONB, nullable=True),
-    sqlalchemy.Column("dead_letter_queue_name", sqlalchemy.String(length=128), nullable=True),
-    sqlalchemy.Column("dead_letter_max_retries", sqlalchemy.Integer, nullable=True),
-    sqlalchemy.Column("dead_letter_min_backoff_seconds", sqlalchemy.Integer, nullable=True),
-    sqlalchemy.Column("dead_letter_max_backoff_seconds", sqlalchemy.Integer, nullable=True),
-    sqlalchemy.Column("created_at", sqlalchemy.DateTime, nullable=False),
-    sqlalchemy.Column("updated_at", sqlalchemy.DateTime, nullable=False),
-)
-
-messages = sqlalchemy.Table(
-    "messages",
-    metadata,
-    sqlalchemy.Column("id", UUID, primary_key=True, nullable=False),
-    sqlalchemy.Column(
-        "queue_id", sqlalchemy.String(length=128), sqlalchemy.ForeignKey("queues.id"), nullable=False
-    ),
-    sqlalchemy.Column("data", JSONB, nullable=False),
-    sqlalchemy.Column("attributes", JSONB, nullable=True),
-    sqlalchemy.Column("delivery_attempts", sqlalchemy.Integer, nullable=False),
-    sqlalchemy.Column("expired_at", sqlalchemy.DateTime, nullable=False),
-    sqlalchemy.Column("scheduled_at", sqlalchemy.DateTime, nullable=False),
-    sqlalchemy.Column("created_at", sqlalchemy.DateTime, nullable=False),
-    sqlalchemy.Column("updated_at", sqlalchemy.DateTime, nullable=False),
-    sqlalchemy.Index("ix_messages_expired_at", "expired_at", postgresql_using="brin"),
-    sqlalchemy.Index("ix_messages_scheduled_at", "scheduled_at", postgresql_using="brin"),
-    sqlalchemy.Index(
-        "ix_messages_expired_at_scheduled_at", "expired_at", "scheduled_at", postgresql_using="brin"
-    ),
-)
+from sqlalchemy.dialects import postgresql
+
+from fastqueue.database import Base
+
+
+class Topic(Base):
+    __tablename__ = "topics"
+
+    id = sqlalchemy.Column(sqlalchemy.String(length=128), primary_key=True, nullable=False)
+    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+
+
+class Queue(Base):
+    __tablename__ = "queues"
+
+    id = sqlalchemy.Column(sqlalchemy.String(length=128), primary_key=True, nullable=False)
+    topic_id = sqlalchemy.Column(
+        sqlalchemy.String(length=128), sqlalchemy.ForeignKey("topics.id"), index=True, nullable=False
+    )
+    ack_deadline_seconds = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
+    message_retention_seconds = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
+    message_filters = sqlalchemy.Column(postgresql.JSONB, nullable=True)
+    dead_letter_queue_name = sqlalchemy.Column(sqlalchemy.String(length=128), nullable=True)
+    dead_letter_max_retries = sqlalchemy.Column(sqlalchemy.Integer, nullable=True)
+    dead_letter_min_backoff_seconds = sqlalchemy.Column(sqlalchemy.Integer, nullable=True)
+    dead_letter_max_backoff_seconds = sqlalchemy.Column(sqlalchemy.Integer, nullable=True)
+    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    updated_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+
+
+class Message(Base):
+    __tablename__ = "messages"
+    __table_args__ = (
+        sqlalchemy.Index("ix_messages_expired_at", "expired_at", postgresql_using="brin"),
+        sqlalchemy.Index("ix_messages_scheduled_at", "scheduled_at", postgresql_using="brin"),
+        sqlalchemy.Index(
+            "ix_messages_expired_at_scheduled_at", "expired_at", "scheduled_at", postgresql_using="brin"
+        ),
+    )
+
+    id = sqlalchemy.Column(postgresql.UUID, primary_key=True, nullable=False)
+    queue_id = sqlalchemy.Column(
+        sqlalchemy.String(length=128), sqlalchemy.ForeignKey("queues.id"), nullable=False
+    )
+    data = sqlalchemy.Column(postgresql.JSONB, nullable=False)
+    attributes = sqlalchemy.Column(postgresql.JSONB, nullable=True)
+    delivery_attempts = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
+    expired_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    scheduled_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    updated_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)

pyproject.toml

@@ -11,7 +11,6 @@ uvicorn = {extras = ["standard"], version = "^0.18.3"}
 SQLAlchemy = "^1.4.40"
 alembic = "^1.8.1"
 psycopg2-binary = "^2.9.3"
-databases = {extras = ["asyncpg"], version = "^0.6.1"}
 typer = {extras = ["all"], version = "^0.6.1"}
 python-json-logger = "^2.0.4"
 

tests/conftest.py

@@ -2,8 +2,28 @@
 from fastapi.testclient import TestClient
 
 from fastqueue.api import app
+from fastqueue.database import engine, SessionLocal
+from fastqueue.models import Base
 
 
 @pytest.fixture
 def client():
     return TestClient(app)
+
+
+@pytest.fixture(scope="module")
+def connection():
+    connection = engine.connect()
+    Base.metadata.create_all(bind=engine)
+    yield connection
+    Base.metadata.drop_all(bind=engine)
+    connection.close()
+
+
+@pytest.fixture(scope="function")
+def session(connection):
+    transaction = connection.begin()
+    session = SessionLocal(bind=connection)
+    yield session
+    session.close()
+    transaction.rollback()