Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

How to query additional databases using cursor in Django Pytests

I am developing a Django app (Django v3.2.10, pytest v7.0.1, pytest-django v4.5.2) which uses cursor to perform raw queries to my secondary DB: my_db2, but when running tests, all the queries return empty results, like if they were running on parallel transactions.

My test file:

@pytest.mark.django_db(transaction=True, databases=['default', 'my_db2'])
class TestItems:
    def test_people(self):
      person1 = PeopleFactory()  # Adds 1 person to my_db2
      assert fetch_all_persons() == 1 # Fails Returns 0

My Factory:

class PeopleFactory(factory.django.DjangoModelFactory):
    id = factory.Sequence(lambda x: x + 1)
    name = factory.Faker('first_name')

    class Meta:
        model = People

My function:

from django.db import connections


def fetch_all_persons():
      with connections['my_db2'].cursor() as cursor:
        cursor.execute(f"SELECT * FROM Persons")
        return len(list(cursor.fetchall())):

According documentation transaction=True should prevent this issue, but it doesn't, does somebody know how to fix it?

Note.- Using the ORM is not an option, this is just a simplified example to represent the issue. The real queries used are way more complex.

like image 445
Ander Avatar asked Apr 24 '26 14:04

Ander


1 Answers

@hoefling and @Arkadiusz Łukasiewicz were right, I just needed to add the corresponding DB within the factories:

class PeopleFactory(factory.django.DjangoModelFactory):
    id = factory.Sequence(lambda x: x + 1)
    name = factory.Faker('first_name')

    class Meta:
        model = People
        database = 'my_db2'

Thank you both.

like image 146
Ander Avatar answered Apr 27 '26 03:04

Ander



Donate For Us

If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!