Dynamic Table creation and mapping

[AsfanUlla]

Hi i am creating a Multi-tenant system where each tenant would have their own separate tables, I was wondering is it possible to create multiple tables using the same schema and map them to the same object. here is a example on what i am trying to achieve in SqlAlchemy https://medium.com/@ketansomvanshi007/dynamic-object-mapping-in-sqlalchemy-for-multitenancy-4e7c5064c816

[dantownsend]

Piccolo lets you specify which schema a table lives in:

class MyTable(Table, schema='schema_1'):
    my_column = Varchar()

If you potentially have many schemas (one for each customer) you can create table classes at run time using this:

piccolo/piccolo/table.py

Lines 1358 to 1387 in 5f972d0

	def create_table_class(
	class_name: str,
	bases: t.Tuple[t.Type] = (Table,),
	class_kwargs: t.Dict[str, t.Any] = {},
	class_members: t.Dict[str, t.Any] = {},
	) -> t.Type[Table]:
	"""
	Used to dynamically create ``Table``subclasses at runtime. Most users
	will not require this. It's mostly used internally for Piccolo's
	migrations.
	:param class_name:
	For example `'MyTable'`.
	:param bases:
	A tuple of parent classes - usually just `(Table,)`.
	:param class_kwargs:
	For example, `{'tablename': 'my_custom_tablename'}`.
	:param class_members:
	For example, `{'my_column': Varchar()}`.
	"""
	return t.cast(
	t.Type[Table],
	types.new_class(
	name=class_name,
	bases=bases,
	kwds=class_kwargs,
	exec_body=lambda namespace: namespace.update(class_members),
	),
	)

For example:

from piccolo.table import create_table_class
from piccolo.columns import Varchar

MyTable = create_table_class(
    class_name='MyTable',
    class_kwargs={
        "schema": 'schema_1',
    },
    class_members={
        'name': Varchar()
    },
)

You can then use that table like any other. For example:

def get_table(schema: str):
    return create_table_class(
        class_name='MyTable',
        class_kwargs={
            "schema": schema,
        },
        class_members={
            'name': Varchar()
        },
    )

async def get_user_data(schema: str):
    MyTable = get_table(schema)
    return await MyTable.objects()

Another solution is to use Piccolo's reflection API.