Proposed PGSchema object

[quoimec]

Hello @olirice! I am back with another issue.

For my use case, I wanted to be able to manage schemas through alembic as well as all of the views, functions etc that I need. To do this, I have thrown together a PGSchema ReplaceableEntity that hooks into your system so I was wondering if this is something you'd be interested in adding to the package as a supported module? Obviously would need a bit of work to tidy it up, make it safer and ensure it's cross-db compatible but just thought I'd offer.

Basic implementation:

class PGSchema(ReplaceableEntity):

    type_ = "schema"

    def __init__(self, schema: str, exists: bool = True):
        super().__init__(
            schema="",
            signature=schema,
            definition=schema
        )
        self.exists = exists

    def to_sql_statement_create(self) -> sa.text:
        return sa.text(f"create schema {self.signature};")

    def to_sql_statement_drop(self, cascade=False) -> sa.text:
        return sa.text(f"drop schema {self.signature};")

    def to_sql_statement_create_or_replace(self) -> sa.text:
        raise NotImplementedError()

    @property
    def identity(self) -> str:
        return f"{self.__class__.__name__}: {self.signature} {self.exists}"

    def to_variable_name(self) -> str:
        return f"schema_{self.signature}"

    def render_self_for_migration(self, omit_definition=False) -> str:
        
        var_name = self.to_variable_name()
        class_name = self.__class__.__name__

        return f"""{var_name} = {class_name}(schema="{self.signature}")\n"""

    @classmethod
    def from_database(cls, sess, schema):
       
        return [PGSchema(schema=x[0]) for x in sess.execute(sa.text("""
            select
                schema_name
            from
                information_schema.schemata 
            where 
                schema_owner != 'rdsadmin' 
                and schema_name != 'public';
        """)).fetchall()]

    def get_database_definition(self, sess, dependencies = None) -> "PGSchema":
        
        exists = sess.execute(sa.text("""
            select exists(
                select 
                    1
                from
                    information_schema.schemata 
                where 
                    schema_owner != 'rdsadmin' 
                    and schema_name = :schema
            );
        """), {"schema": self.signature}).fetchone()[0]

        if exists:
            return self
        else:
            return PGSchema(schema=self.schema, exists=False)

[olirice]

hi again,

Yes, I am interested in adding schema creation/dropping support. Maintaining schemas in migrations is typically low effort since they only need to be created once and don't require updates. For that reason, i haven't invested the time to do it yet

thank you for the reference implementation