-
-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
4 changed files
with
257 additions
and
42 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,43 +1,3 @@ | ||
from typing import Any | ||
from .generics import DeclarativeMixin | ||
|
||
from pydantic import BaseModel, ConfigDict | ||
from sqlalchemy.orm import Mapped, relationship | ||
|
||
|
||
class DeclarativeMixin(BaseModel): | ||
""" | ||
Mixin for declarative base models. | ||
""" | ||
|
||
model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True) | ||
|
||
@classmethod | ||
def declarative(cls) -> Any: | ||
return cls.generate_model_declarative() | ||
|
||
@classmethod | ||
def generate_model_declarative(cls) -> Any: | ||
""" | ||
Transforms a core Saffier table into a Declarative model table. | ||
""" | ||
Base = cls.meta.registry.declarative_base | ||
|
||
# Build the original table | ||
fields = {"__table__": cls.table} | ||
|
||
# Generate base | ||
model_table = type(cls.__name__, (Base,), fields) | ||
|
||
# Make sure if there are foreignkeys, builds the relationships | ||
for column in cls.table.columns: | ||
if not column.foreign_keys: | ||
continue | ||
|
||
# Maps the relationships with the foreign keys and related names | ||
field = cls.fields.get(column.name) | ||
mapped_model: Mapped[field.to.__name__] = relationship(field.to.__name__) | ||
|
||
# Adds to the current model | ||
model_table.__mapper__.add_property(f"{column.name}_relation", mapped_model) | ||
|
||
return model_table | ||
__all__ = ["DeclarativeMixin"] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,43 @@ | ||
from typing import Any | ||
|
||
from pydantic import BaseModel, ConfigDict | ||
from sqlalchemy.orm import Mapped, relationship | ||
|
||
|
||
class DeclarativeMixin(BaseModel): | ||
""" | ||
Mixin for declarative base models. | ||
""" | ||
|
||
model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True) | ||
|
||
@classmethod | ||
def declarative(cls) -> Any: | ||
return cls.generate_model_declarative() | ||
|
||
@classmethod | ||
def generate_model_declarative(cls) -> Any: | ||
""" | ||
Transforms a core Saffier table into a Declarative model table. | ||
""" | ||
Base = cls.meta.registry.declarative_base | ||
|
||
# Build the original table | ||
fields = {"__table__": cls.table} | ||
|
||
# Generate base | ||
model_table = type(cls.__name__, (Base,), fields) | ||
|
||
# Make sure if there are foreignkeys, builds the relationships | ||
for column in cls.table.columns: | ||
if not column.foreign_keys: | ||
continue | ||
|
||
# Maps the relationships with the foreign keys and related names | ||
field = cls.fields.get(column.name) | ||
mapped_model: Mapped[field.to.__name__] = relationship(field.to.__name__) | ||
|
||
# Adds to the current model | ||
model_table.__mapper__.add_property(f"{column.name}_relation", mapped_model) | ||
|
||
return model_table |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,54 @@ | ||
import pytest | ||
from tests.settings import DATABASE_URL | ||
|
||
import edgy | ||
from edgy.testclient import DatabaseTestClient as Database | ||
|
||
pytestmark = pytest.mark.anyio | ||
|
||
database = Database(DATABASE_URL) | ||
models = edgy.Registry(database=database) | ||
|
||
|
||
class User(edgy.Model): | ||
id = edgy.IntegerField(primary_key=True) | ||
name = edgy.CharField(max_length=100) | ||
|
||
class Meta: | ||
registry = models | ||
|
||
|
||
@pytest.fixture(autouse=True, scope="module") | ||
async def create_test_database(): | ||
await models.create_all() | ||
yield | ||
await models.drop_all() | ||
|
||
|
||
@pytest.fixture(autouse=True) | ||
async def rollback_connections(): | ||
with database.force_rollback(): | ||
async with database: | ||
yield | ||
|
||
|
||
async def test_meta_tablename(): | ||
await User.query.create(name="edgy") | ||
users = await User.query.all() | ||
|
||
assert len(users) == 1 | ||
|
||
user = await User.query.get(name="edgy") | ||
|
||
assert user.meta.tablename == "users" | ||
|
||
|
||
async def test_meta_registry(): | ||
await User.query.create(name="edgy") | ||
users = await User.query.all() | ||
|
||
assert len(users) == 1 | ||
|
||
user = await User.query.get(name="edgy") | ||
|
||
assert user.meta.registry == models |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,158 @@ | ||
from typing import ClassVar | ||
|
||
import pytest | ||
from tests.settings import DATABASE_URL | ||
|
||
import edgy | ||
from edgy import Manager, QuerySet | ||
from edgy.exceptions import ForeignKeyBadConfigured, ImproperlyConfigured | ||
from edgy.testclient import DatabaseTestClient as Database | ||
|
||
pytestmark = pytest.mark.anyio | ||
|
||
database = Database(DATABASE_URL) | ||
models = edgy.Registry(database=database) | ||
|
||
|
||
class User(edgy.Model): | ||
id = edgy.IntegerField(primary_key=True) | ||
name = edgy.CharField(max_length=100) | ||
|
||
class Meta: | ||
registry = models | ||
|
||
|
||
class ObjectsManager(Manager): | ||
def get_queryset(self) -> QuerySet: | ||
queryset = super().get_queryset().filter(name__icontains="a") | ||
return queryset | ||
|
||
|
||
async def test_improperly_configured_for_multiple_managers_on_abstract_class(): | ||
with pytest.raises(ImproperlyConfigured) as raised: | ||
|
||
class BaseModel(edgy.Model): | ||
query: ClassVar[Manager] = ObjectsManager() | ||
languages: ClassVar[Manager] = ObjectsManager() | ||
|
||
class Meta: | ||
abstract = True | ||
registry = models | ||
|
||
assert raised.value.args[0] == "Multiple managers are not allowed in abstract classes." | ||
|
||
|
||
async def test_improperly_configured_for_primary_key(): | ||
with pytest.raises(ImproperlyConfigured) as raised: | ||
|
||
class BaseModel(edgy.Model): | ||
id = edgy.IntegerField(primary_key=False) | ||
query: ClassVar[Manager] = ObjectsManager() | ||
languages: ClassVar[Manager] = ObjectsManager() | ||
|
||
class Meta: | ||
registry = models | ||
|
||
assert ( | ||
raised.value.args[0] | ||
== "Cannot create model BaseModel without explicit primary key if field 'id' is already present." | ||
) | ||
|
||
|
||
async def test_improperly_configured_for_multiple_primary_keys(): | ||
with pytest.raises(ImproperlyConfigured) as raised: | ||
|
||
class BaseModel(edgy.Model): | ||
name = edgy.IntegerField(primary_key=True) | ||
query: ClassVar[Manager] = ObjectsManager() | ||
languages: ClassVar[Manager] = ObjectsManager() | ||
|
||
class Meta: | ||
registry = models | ||
|
||
assert raised.value.args[0] == "Cannot create model BaseModel with multiple primary keys." | ||
|
||
|
||
@pytest.mark.parametrize("_type,value", [("int", 1), ("dict", {"name": "test"}), ("set", set())]) | ||
async def test_improperly_configured_for_unique_together(_type, value): | ||
with pytest.raises(ImproperlyConfigured) as raised: | ||
|
||
class BaseModel(edgy.Model): | ||
name = edgy.IntegerField() | ||
query: ClassVar[Manager] = ObjectsManager() | ||
languages: ClassVar[Manager] = ObjectsManager() | ||
|
||
class Meta: | ||
registry = models | ||
unique_together = value | ||
|
||
assert raised.value.args[0] == f"unique_together must be a tuple or list. Got {_type} instead." | ||
|
||
|
||
@pytest.mark.parametrize( | ||
"value", | ||
[(1, dict), ["str", 1, set], [1], [dict], [set], [set, dict, list, tuple]], | ||
ids=[ | ||
"int-and-dict", | ||
"str-int-set", | ||
"list-of-int", | ||
"list-of-dict", | ||
"list-of-set", | ||
"list-of-set-dict-tuple-and-lists", | ||
], | ||
) | ||
async def test_value_error_for_unique_together(value): | ||
with pytest.raises(ValueError) as raised: | ||
|
||
class BaseModel(edgy.Model): | ||
name = edgy.IntegerField() | ||
query: ClassVar[Manager] = ObjectsManager() | ||
languages: ClassVar[Manager] = ObjectsManager() | ||
|
||
class Meta: | ||
registry = models | ||
unique_together = value | ||
|
||
assert ( | ||
raised.value.args[0] | ||
== "The values inside the unique_together must be a string, a tuple of strings or an instance of UniqueConstraint." | ||
) | ||
|
||
|
||
def test_raises_value_error_on_wrong_type(): | ||
with pytest.raises(ValueError) as raised: | ||
|
||
class User(edgy.Model): | ||
name = edgy.CharField(max_length=255) | ||
|
||
class Meta: | ||
registry = models | ||
indexes = ["name"] | ||
|
||
assert raised.value.args[0] == "Meta.indexes must be a list of Index types." | ||
|
||
|
||
def test_raises_ForeignKeyBadConfigured(): | ||
name = "profiles" | ||
|
||
with pytest.raises(ForeignKeyBadConfigured) as raised: | ||
|
||
class User(edgy.Model): | ||
name = edgy.CharField(max_length=255) | ||
|
||
class Meta: | ||
registry = models | ||
|
||
class Profile(edgy.Model): | ||
user = edgy.ForeignKey(User, null=False, on_delete=edgy.CASCADE, related_name=name) | ||
another_user = edgy.ForeignKey( | ||
User, null=False, on_delete=edgy.CASCADE, related_name=name | ||
) | ||
|
||
class Meta: | ||
registry = models | ||
|
||
assert ( | ||
raised.value.args[0] | ||
== f"Multiple related_name with the same value '{name}' found to the same target. Related names must be different." | ||
) |