feat: Allow globbing table names to select subsets of tables.

docs/source/database.rst

@@ -311,9 +311,6 @@ This can be a great way to keep track of all the tables a given block of code in
    )
 
 
-As you can see, in the above example, tables accepts _any_ of: the string table name, the
-SQLAlchemy table object, or a SQLAlchemy model class.
-
 .. code-block:: python
 
    # tests/test_something.py:
@@ -326,6 +323,24 @@ SQLAlchemy table object, or a SQLAlchemy model class.
        assert ["ABCDE"] == result
 
 
+The :code:`tables` argument accepts any of:
+
+* SQLAlchemy declarative model class
+* SQLAlchemy table object
+* Exact string table name
+* Globbed table name
+
+  Globbing, in comparison to regular expressions, in this context tends to lead to shorter
+  and easier to read definitions. This is especially true when one uses schemas, leading
+  to :code:`.` literals in your fully qualified table names.
+
+  .. code-block:: python
+
+     create_<backend>_fixture(Base, tables=['schema.*'])  # Only tables for a specific schema
+     create_<backend>_fixture(Base, tables=['category_*'])  # Only tables with a specific suffix
+     create_<backend>_fixture(Base, tables=['*_category'])  # Only tables with a specific prefix
+
+
 Rows
 ~~~~
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pytest-mock-resources"
-version = "1.3.2"
+version = "1.4.0"
 description = "A pytest plugin for easily instantiating reproducible mock resources."
 authors = [
     "Omar Khan <oakhan3@gmail.com>",

src/pytest_mock_resources/fixture/database/relational/generic.py

@@ -1,4 +1,5 @@
 import abc
+import fnmatch
 
 import attr
 import six
@@ -114,24 +115,11 @@ def _create_tables(self, metadata):
             metadata.create_all(self.engine)
             return
 
-        table_objects = set()
-        for table in self.tables:
-            if isinstance(table, DeclarativeMeta):
-                table_objects.add(table.__table__)
-            elif isinstance(table, Table):
-                table_objects.add(table)
-            else:
-                table_name = table
-
-                if table_name not in metadata.tables:
-                    raise ValueError(
-                        'Could not identify table "{}" from: {}'.format(
-                            table_name, ", ".join(sorted(metadata.tables.keys()))
-                        )
-                    )
-
-                table = metadata.tables[table_name]
-                table_objects.add(table)
+        table_objects = {
+            table_object
+            for table in self.tables
+            for table_object in identify_matching_tables(metadata, table)
+        }
 
         metadata.create_all(self.engine, tables=list(table_objects))
 
@@ -167,3 +155,25 @@ def manage(self, session=None):
                 yield self.engine
         finally:
             self.engine.dispose()
+
+
+def identify_matching_tables(metadata, table_specifier):
+    if isinstance(table_specifier, DeclarativeMeta):
+        return [table_specifier.__table__]
+
+    if isinstance(table_specifier, Table):
+        return [table_specifier]
+
+    tables = [
+        table
+        for table_name, table in metadata.tables.items()
+        if fnmatch.fnmatch(table_name, table_specifier)
+    ]
+
+    if tables:
+        return tables
+
+    table_names = ", ".join(sorted(metadata.tables.keys()))
+    raise ValueError(
+        'Could not identify any tables matching "{}" from: {}'.format(table_specifier, table_names)
+    )

tests/fixture/database/relational/test_generic.py

@@ -1,10 +1,11 @@
 import pytest
 import sqlalchemy
-from sqlalchemy import Column, Integer, SmallInteger, Unicode
+from sqlalchemy import Column, Integer, MetaData, SmallInteger, Table, Unicode
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 
 from pytest_mock_resources import create_postgres_fixture, create_sqlite_fixture, Rows
+from pytest_mock_resources.fixture.database.relational.generic import identify_matching_tables
 
 Base = declarative_base()
 
@@ -39,6 +40,7 @@ class Other2(Base):
 sqlite_duplicate = create_sqlite_fixture(
     Base, tables=[Thing, Other2.__table__, "thing", "other.other"]
 )
+sqlite_glob = create_sqlite_fixture(Base, tables=["*.other"])
 
 
 class TestTablesArg:
@@ -72,6 +74,13 @@ def test_table_duplicate(self, sqlite_duplicate):
         with pytest.raises(sqlalchemy.exc.OperationalError):
             sqlite_duplicate.execute("select * from public.other")
 
+    def test_glob(self, sqlite_glob):
+        with pytest.raises(sqlalchemy.exc.OperationalError):
+            sqlite_glob.execute("select * from thing")
+
+        sqlite_glob.execute("select * from other.other")
+        sqlite_glob.execute("select * from public.other")
+
 
 PGBase = declarative_base()
 
@@ -135,3 +144,34 @@ def test_session2(self, sqlite2):
     def test_session_pg(self, pg_session):
         result = pg_session.query(Quarter).one()
         assert result.id == 1
+
+
+class Test__identify_matching_tables:
+    @staticmethod
+    def by_tablename(table):
+        return table.schema + table.name
+
+    def setup(self):
+        self.metadata = metadata = MetaData()
+        self.base = declarative_base()
+
+        self.one_foo = Table("foo", metadata, schema="one")
+        self.one_foo_bar = Table("foo_bar", metadata, schema="one")
+        self.two_foo = Table("foo", metadata, schema="two")
+        self.two_far = Table("far", metadata, schema="two")
+
+    def test_no_glob(self):
+        result = identify_matching_tables(self.metadata, "one.foo")
+        assert sorted(result, key=self.by_tablename) == [self.one_foo]
+
+    def test_glob_table_on_schema(self):
+        result = identify_matching_tables(self.metadata, "one.*")
+        assert sorted(result, key=self.by_tablename) == [self.one_foo, self.one_foo_bar]
+
+    def test_glob_schema_on_table(self):
+        result = identify_matching_tables(self.metadata, "*.foo")
+        assert sorted(result, key=self.by_tablename) == [self.one_foo, self.two_foo]
+
+    def test_glob_optional_char(self):
+        result = identify_matching_tables(self.metadata, "two.f??")
+        assert sorted(result, key=self.by_tablename) == [self.two_far, self.two_foo]