Added table, image and file fields and some code refactoring

Author: eadwinCode

README.md

@@ -10,11 +10,8 @@
 
 ## Project Status
 - 70% done
-- SQLAlchemy Table support with ModelSession
-- Migration custom revision directives
+- tests
 - Documentation
-- File Field
-- Image Field
 
 ## Introduction
 Ellar SQLAlchemy Module simplifies the integration of SQLAlchemy and Alembic migration tooling into your ellar application.

ellar_sqlalchemy/constant.py

@@ -6,6 +6,14 @@
 TABLE_KEY = "__table__"
 ABSTRACT_KEY = "__abstract__"
 
+NAMING_CONVERSION = {
+    "ix": "ix_%(column_0_label)s",
+    "uq": "uq_%(table_name)s_%(column_0_name)s",
+    "ck": "ck_%(table_name)s_%(constraint_name)s",
+    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
+    "pk": "pk_%(table_name)s",
+}
+
 
 class DeclarativeBasePlaceHolder(sa_orm.DeclarativeBase):
     pass

ellar_sqlalchemy/migrations/base.py

@@ -18,6 +18,18 @@ def __init__(self, db_service: EllarSQLAlchemyService) -> None:
         self.db_service = db_service
         self.use_two_phase = db_service.migration_options.use_two_phase
 
+    def get_user_context_configurations(self) -> t.Dict[str, t.Any]:
+        conf_args = dict(self.db_service.migration_options.context_configure)
+
+        # detecting type changes
+        conf_args.setdefault("compare_type", True)
+        conf_args.setdefault("render_as_batch", True)
+        # If you want to ignore things like these, set the following as a class attribute
+        # __table_args__ = {"info": {"skip_autogen": True}}
+        conf_args.setdefault("include_object", self.include_object)
+        conf_args.setdefault("dialect_opts", {"paramstyle": "named"})
+        return conf_args
+
     def include_object(
         self,
         obj: SchemaItem,

ellar_sqlalchemy/migrations/multiple.py

@@ -96,6 +96,8 @@ def run_migrations_offline(self, context: "EnvironmentContext") -> None:
         # for --sql use case, run migrations for each URL into
         # individual files.
 
+        conf_args = self.get_user_context_configurations()
+
         for key, engine in self.db_service.engines.items():
             logger.info("Migrating database %s" % key)
 
@@ -104,18 +106,14 @@ def run_migrations_offline(self, context: "EnvironmentContext") -> None:
 
             file_ = "%s.sql" % key
             logger.info("Writing output to %s" % file_)
+
             with open(file_, "w") as buffer:
                 context.configure(
                     url=url,
                     output_buffer=buffer,
                     target_metadata=metadata,
                     literal_binds=True,
-                    dialect_opts={"paramstyle": "named"},
-                    # If you want to ignore things like these, set the following as a class attribute
-                    # __table_args__ = {"info": {"skip_autogen": True}}
-                    include_object=self.include_object,
-                    # detecting type changes
-                    # compare_type=True,
+                    **conf_args,
                 )
                 with context.begin_transaction():
                     context.run_migrations(engine_name=key)
@@ -126,12 +124,11 @@ def _migration_action(
         # this callback is used to prevent an auto-migration from being generated
         # when there are no changes to the schema
         # reference: http://alembic.zzzcomputing.com/en/latest/cookbook.html
-        conf_args = {
-            "process_revision_directives": self.default_process_revision_directives
-        }
-        # conf_args = current_app.extensions['migrate'].configure_args
-        # if conf_args.get("process_revision_directives") is None:
-        #     conf_args["process_revision_directives"] = process_revision_directives
+
+        conf_args = self.get_user_context_configurations()
+        conf_args.setdefault(
+            "process_revision_directives", self.default_process_revision_directives
+        )
 
         try:
             for db_info in db_infos:

ellar_sqlalchemy/migrations/single.py

@@ -53,16 +53,13 @@ def run_migrations_offline(self, context: "EnvironmentContext") -> None:
         key, engine = self.db_service.engines.popitem()
         metadata = get_database_bind(key, certain=True)
 
+        conf_args = self.get_user_context_configurations()
+
         context.configure(
             url=str(engine.url).replace("%", "%%"),
             target_metadata=metadata,
             literal_binds=True,
-            dialect_opts={"paramstyle": "named"},
-            # If you want to ignore things like these, set the following as a class attribute
-            # __table_args__ = {"info": {"skip_autogen": True}}
-            include_object=self.include_object,
-            # detecting type changes
-            # compare_type=True,
+            **conf_args,
         )
 
         with context.begin_transaction():
@@ -77,12 +74,10 @@ def _migration_action(
         # this callback is used to prevent an auto-migration from being generated
         # when there are no changes to the schema
         # reference: http://alembic.zzzcomputing.com/en/latest/cookbook.html
-        conf_args = {
-            "process_revision_directives": self.default_process_revision_directives
-        }
-        # conf_args = current_app.extensions['migrate'].configure_args
-        # if conf_args.get("process_revision_directives") is None:
-        #     conf_args["process_revision_directives"] = process_revision_directives
+        conf_args = self.get_user_context_configurations()
+        conf_args.setdefault(
+            "process_revision_directives", self.default_process_revision_directives
+        )
 
         context.configure(connection=connection, target_metadata=metadata, **conf_args)
 

ellar_sqlalchemy/model/__init__.py

@@ -1,9 +1,11 @@
 from .base import Model
+from .table import Table
 from .typeDecorator import GUID, GenericIP
 from .utils import make_metadata
 
 __all__ = [
     "Model",
+    "Table",
     "make_metadata",
     "GUID",
     "GenericIP",

ellar_sqlalchemy/model/base.py

@@ -1,13 +1,14 @@
 import types
 import typing as t
 
+import sqlalchemy as sa
 import sqlalchemy.orm as sa_orm
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import DeclarativeBase
 
 from ellar_sqlalchemy.constant import (
     DATABASE_BIND_KEY,
     DEFAULT_KEY,
+    NAMING_CONVERSION,
 )
 
 from .database_binds import get_database_bind, has_database_bind, update_database_binds
@@ -70,7 +71,7 @@ def get_session(cls: t.Type[Model]) -> None:
         return SQLAlchemyDefaultBase
 
 
-class ModelMeta(type(DeclarativeBase)):  # type:ignore[misc]
+class ModelMeta(type(sa_orm.DeclarativeBase)):  # type:ignore[misc]
     def __new__(
         mcs,
         name: str,
@@ -92,11 +93,20 @@ def __new__(
 
         if not skip_default_base_check:
             if SQLAlchemyDefaultBase is None:
-                raise Exception(
-                    "EllarSQLAlchemy Default Declarative Base has not been configured."
-                    "\nPlease call `configure_model_declarative_base` before ORM Model construction"
-                    " or Use EllarSQLAlchemy Service"
+                # raise Exception(
+                #     "EllarSQLAlchemy Default Declarative Base has not been configured."
+                #     "\nPlease call `configure_model_declarative_base` before ORM Model construction"
+                #     " or Use EllarSQLAlchemy Service"
+                # )
+                _model_as_base(
+                    "SQLAlchemyDefaultBase",
+                    (),
+                    {
+                        "skip_default_base_check": True,
+                        "metadata": sa.MetaData(naming_convention=NAMING_CONVERSION),
+                    },
                 )
+                _bases = [SQLAlchemyDefaultBase, *_bases]
             elif SQLAlchemyDefaultBase and SQLAlchemyDefaultBase not in _bases:
                 _bases = [SQLAlchemyDefaultBase, *_bases]
 

ellar_sqlalchemy/model/table.py

@@ -0,0 +1,63 @@
+import typing as t
+
+import sqlalchemy as sa
+import sqlalchemy.sql.schema as sa_sql_schema
+
+from ellar_sqlalchemy.constant import DEFAULT_KEY
+from ellar_sqlalchemy.model import make_metadata
+
+
+class Table(sa.Table):
+    """
+    Custom SQLAlchemy Table class that supports database-binding
+    E.g.:
+
+        user_book_m2m = Table(
+            "user_book",
+            sa.Column("user_id", sa.ForeignKey(User.id), primary_key=True),
+            sa.Column("book_id", sa.ForeignKey(Book.id), primary_key=True),
+            __database__='default'
+        )
+    """
+
+    @t.overload
+    def __init__(
+        self,
+        name: str,
+        *args: sa_sql_schema.SchemaItem,
+        __database__: t.Optional[str] = None,
+        **kwargs: t.Any,
+    ) -> None:
+        ...
+
+    @t.overload
+    def __init__(
+        self,
+        name: str,
+        metadata: sa.MetaData,
+        *args: sa_sql_schema.SchemaItem,
+        **kwargs: t.Any,
+    ) -> None:
+        ...
+
+    @t.overload
+    def __init__(
+        self, name: str, *args: sa_sql_schema.SchemaItem, **kwargs: t.Any
+    ) -> None:
+        ...
+
+    def __init__(
+        self, name: str, *args: sa_sql_schema.SchemaItem, **kwargs: t.Any
+    ) -> None:
+        super().__init__(name, *args, **kwargs)  # type: ignore[arg-type]
+
+    def __new__(
+        cls, *args: t.Any, __database__: t.Optional[str] = None, **kwargs: t.Any
+    ) -> "Table":
+        # If a metadata arg is passed, go directly to the base Table. Also do
+        # this for no args so the correct error is shown.
+        if not args or (len(args) >= 2 and isinstance(args[1], sa.MetaData)):
+            return super().__new__(cls, *args, **kwargs)
+
+        metadata = make_metadata(__database__ or DEFAULT_KEY)
+        return super().__new__(cls, *[args[0], metadata, *args[1:]], **kwargs)

ellar_sqlalchemy/model/typeDecorator/__init__.py

@@ -1,13 +1,14 @@
-# from .file import FileField
+from .file import FileField, FileFieldBase, FileObject
 from .guid import GUID
+from .image import CroppingDetails, ImageFileField
 from .ipaddress import GenericIP
 
-# from .image import CroppingDetails, ImageFileField
-
 __all__ = [
     "GUID",
     "GenericIP",
-    # "CroppingDetails",
-    # "FileField",
-    # "ImageFileField",
+    "CroppingDetails",
+    "FileField",
+    "ImageFileField",
+    "FileObject",
+    "FileFieldBase",
 ]

…/model/typeDecorator/exceptions.py.ellar …lchemy/model/typeDecorator/exceptions.pyellar_sqlalchemy/model/typeDecorator/exceptions.py.ellar renamed to ellar_sqlalchemy/model/typeDecorator/exceptions.py ellar_sqlalchemy/model/typeDecorator/exceptions.py.ellar renamed to ellar_sqlalchemy/model/typeDecorator/exceptions.py

@@ -1,6 +1,12 @@
-class ContentTypeValidationError(Exception):
-    def __init__(self, content_type=None, valid_content_types=None):
+import typing as t
+
 
+class ContentTypeValidationError(Exception):
+    def __init__(
+        self,
+        content_type: t.Optional[str] = None,
+        valid_content_types: t.Optional[t.List[str]] = None,
+    ) -> None:
         if content_type is None:
             message = "Content type is not provided. "
         else:
@@ -21,5 +27,5 @@ class InvalidImageOperationError(Exception):
 
 
 class MaximumAllowedFileLengthError(Exception):
-    def __init__(self, max_length: int):
+    def __init__(self, max_length: int) -> None:
         super().__init__("Cannot store files larger than: %d bytes" % max_length)