diff --git a/example/config-app/uv.lock b/example/config-app/uv.lock
index 6868b878..3923cfba 100644
--- a/example/config-app/uv.lock
+++ b/example/config-app/uv.lock
@@ -162,7 +162,7 @@ wheels = [
 
 [[package]]
 name = "fastapi-startkit"
-version = "0.24.0"
+version = "0.25.0"
 source = { editable = "../../fastapi_startkit" }
 dependencies = [
     { name = "cleo" },
diff --git a/example/database-app/uv.lock b/example/database-app/uv.lock
index f2618dee..bcd11761 100644
--- a/example/database-app/uv.lock
+++ b/example/database-app/uv.lock
@@ -498,7 +498,7 @@ wheels = [
 
 [[package]]
 name = "fastapi-startkit"
-version = "0.24.0"
+version = "0.25.0"
 source = { editable = "../../fastapi_startkit" }
 dependencies = [
     { name = "cleo" },
diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/__init__.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/__init__.py
index 731e8c69..b9117281 100644
--- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/__init__.py
+++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/__init__.py
@@ -3,6 +3,30 @@
 from .migrations.Migration import Migration
 from .migrations.Migrator import Migrator
 from .models import Model
+from .models.fields import CreatedAtField, DateTimeField, Field, ModelField, UpdatedAtField
 from .providers import DatabaseProvider
+from .relationships import BelongsTo, BelongsToMany, HasMany, HasManyThrough, HasOne, HasOneThrough, MorphTo
 
-__all__ = ["DatabaseProvider", "PostgresConfig", "MySQLConfig", "SQLiteConfig", "Model", "DB", "Migration", "Migrator"]
+__all__ = [
+    "DatabaseProvider",
+    "PostgresConfig",
+    "MySQLConfig",
+    "SQLiteConfig",
+    "Model",
+    "DB",
+    "Migration",
+    "Migrator",
+    "ModelField",
+    "DateTimeField",
+    "CreatedAtField",
+    "UpdatedAtField",
+    "Field",
+    # Relationships
+    "HasOne",
+    "BelongsTo",
+    "HasMany",
+    "HasManyThrough",
+    "BelongsToMany",
+    "HasOneThrough",
+    "MorphTo",
+]
diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/commands/stubs/table_migration.stub b/fastapi_startkit/src/fastapi_startkit/masoniteorm/commands/stubs/table_migration.stub
index 008964ce..907c3e52 100644
--- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/commands/stubs/table_migration.stub
+++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/commands/stubs/table_migration.stub
@@ -1,6 +1,6 @@
 """__MIGRATION_NAME__ Migration."""
 
-from fastapi_startkit.masoniteorm.migrations import Migration
+from fastapi_startkit.masoniteorm import Migration
 
 
 class __MIGRATION_NAME__(Migration):
diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/attribute.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/attribute.py
index ba774ce4..fd59ae18 100644
--- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/attribute.py
+++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/attribute.py
@@ -105,4 +105,7 @@ def get_dirty(self) -> dict:
         }
 
     def get_attributes_for_insert(self) -> dict:
-        return {**self._attributes, **self._dirty_attributes}
+        # _dirty_attributes already went through set_attribute (casts applied on assignment).
+        # _attributes is set raw via new_model_instance, so apply set casts here.
+        casted = {k: self.caster.set(k, v) for k, v in self._attributes.items()}
+        return {**casted, **self._dirty_attributes}
diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/caster.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/caster.py
index a78c4e7d..3153cc08 100644
--- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/caster.py
+++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/caster.py
@@ -111,6 +111,66 @@ def set(self, value):
         return str(value)
 
 
+class TimeCast(BaseCast):
+    """Casts a value to datetime.time; stored as HH:MM:SS string"""
+
+    def get(self, value):
+        if not value:
+            return None
+        if isinstance(value, datetime.time):
+            return value
+        return datetime.time.fromisoformat(str(value))
+
+    def set(self, value):
+        if not value:
+            return None
+        if isinstance(value, datetime.time):
+            return value.strftime("%H:%M:%S")
+        return str(value)
+
+
+class TimeDeltaCast(BaseCast):
+    """Casts a value to datetime.timedelta; stored as total seconds"""
+
+    def get(self, value):
+        if value is None:
+            return None
+        if isinstance(value, datetime.timedelta):
+            return value
+        return datetime.timedelta(seconds=float(value))
+
+    def set(self, value):
+        if value is None:
+            return None
+        if isinstance(value, datetime.timedelta):
+            return value.total_seconds()
+        return float(value)
+
+
+@dataclass
+class ModelCast(BaseCast):
+    model_class: type = field(default=None)
+
+    def get(self, value):
+        if value is None:
+            return None
+        if isinstance(value, self.model_class):
+            return value
+        data = json.loads(value) if isinstance(value, str) else value
+        return self.model_class(**data)
+
+    def set(self, value) -> Optional[str]:
+        if value is None:
+            return None
+        if isinstance(value, self.model_class):
+            if hasattr(value, "model_dump_json"):
+                return value.model_dump_json()
+            return json.dumps(value.__dict__)
+        if isinstance(value, dict):
+            return json.dumps(value)
+        return value
+
+
 class Caster:
     casts = {}
 
@@ -121,6 +181,8 @@ class Caster:
         "float": FloatCast,
         "date": DateCast,
         "decimal": DecimalCast,
+        "time": TimeCast,
+        "timedelta": TimeDeltaCast,
     }
 
     IGNORE_CASTS = ["caster", "db_manager"]
@@ -156,21 +218,26 @@ def build_casts(cls, model):
         annotations = {
             k: v for k, v in annotations.items() if k not in cls.IGNORE_CASTS
         }
-        from .fields import FieldDescriptor
+        from .fields import ModelField, FieldDescriptor
 
         # 1. Collect all potential fields (annotations + descriptors)
         all_field_names = set(annotations.keys())
         descriptors = {}
-        for name, attr in cls.__dict__.items():
-            if isinstance(attr, FieldDescriptor):
+        for name, attr in model.__dict__.items():
+            if isinstance(attr, (FieldDescriptor, ModelField)):
                 all_field_names.add(name)
                 descriptors[name] = attr
 
         casts = {}
         for field_name in all_field_names:
-            # 2. Get Type Hint and FieldInfo
             typ = annotations.get(field_name) or "str"
             descriptor = descriptors.get(field_name, None)
+
+            # AttributeField: use the type annotation as the model class
+            if isinstance(descriptor, ModelField):
+                casts[field_name] = ModelCast(model_class=typ)
+                continue
+
             field_info = (
                 descriptor.field_info
                 if isinstance(descriptor, FieldDescriptor)
@@ -204,12 +271,29 @@ def normalize_type(t):
             or t is Carbon
         ):
             return "date"
+        if t is datetime.time:
+            return "time"
+        if t is datetime.timedelta:
+            return "timedelta"
         if isinstance(t, type):
             if issubclass(t, Enum) or hasattr(t, "get") or hasattr(t, "set"):
                 return t
 
         return "str"
 
+    @staticmethod
+    def _apply_default(cast: "BaseCast"):
+        """Return the Field default/default_factory value, or None if none is set."""
+        from pydantic_core import PydanticUndefined
+
+        if cast.config is None:
+            return None
+        if cast.config.default is not PydanticUndefined:
+            return cast.config.default
+        if cast.config.default_factory is not None:
+            return cast.config.default_factory()
+        return None
+
     def get(self, attribute: str, value: Any) -> Any:
         if attribute not in self.casts:
             return value
@@ -220,7 +304,10 @@ def get(self, attribute: str, value: Any) -> Any:
             return str(value) if value is not None else None
 
         if isinstance(cast, BaseCast):
-            return cast.get(value)
+            result = cast.get(value)
+            if result is None:
+                result = self._apply_default(cast)
+            return result
 
         if isinstance(cast, type):
             if issubclass(cast, Enum):
diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/fields.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/fields.py
index 52ea7f35..4140dc6b 100644
--- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/fields.py
+++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/fields.py
@@ -41,6 +41,19 @@ def Field(*args, **kwargs) -> Any:
     return FieldDescriptor(BaseField(*args, **kwargs))
 
 
+class ModelField:
+    def __set_name__(self, owner, name):
+        self.name = name
+
+    def __get__(self, instance, owner):
+        if instance is None:
+            return self
+        return instance.get_attribute(self.name)
+
+    def __set__(self, instance, value):
+        instance.set_attribute(self.name, value)
+
+
 class DateTimeField:
     def __init__(self, fmt: str = "YYYY-MM-DD HH:mm:ss", tz: str = "UTC"):
         self.format = fmt
diff --git a/fastapi_startkit/src/fastapi_startkit/storage/__init__.py b/fastapi_startkit/src/fastapi_startkit/storage/__init__.py
index dc72a75d..b9a061b4 100644
--- a/fastapi_startkit/src/fastapi_startkit/storage/__init__.py
+++ b/fastapi_startkit/src/fastapi_startkit/storage/__init__.py
@@ -1,3 +1,6 @@
 from .storage import Storage
 from .config import S3Config, LocalDiskConfig, PublicDiskConfig
 from .drivers.fake import FakeDriver
+from .providers.provider import StorageProvider
+
+__all__ = ["Storage", "S3Config", "LocalDiskConfig", "PublicDiskConfig", "FakeDriver", "StorageProvider"]
diff --git a/fastapi_startkit/tests/masoniteorm/fixtures/casts.py b/fastapi_startkit/tests/masoniteorm/fixtures/casts.py
new file mode 100644
index 00000000..c06b74df
--- /dev/null
+++ b/fastapi_startkit/tests/masoniteorm/fixtures/casts.py
@@ -0,0 +1,9 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+class Address(BaseModel):
+    address: Optional[str] = None
+    city: Optional[str] = None
+    state: Optional[str] = None
+    country: Optional[str] = None
diff --git a/fastapi_startkit/tests/masoniteorm/fixtures/migration.py b/fastapi_startkit/tests/masoniteorm/fixtures/migration.py
index 454d41b4..87c27c63 100644
--- a/fastapi_startkit/tests/masoniteorm/fixtures/migration.py
+++ b/fastapi_startkit/tests/masoniteorm/fixtures/migration.py
@@ -15,6 +15,11 @@ async def migrate(schema: Schema) -> None:
         table.string("email").unique()
         table.boolean("is_admin").default(False)
         table.timestamp("email_verified_at").nullable()
+        table.date("date_of_birth").nullable()
+        table.decimal("session_duration").nullable()
+        table.string("punch_in_time").nullable()
+        table.json("preferences").nullable()
+        table.text("address").nullable()
         table.timestamps()
 
     async with await schema.create_table_if_not_exists("profiles") as table:
diff --git a/fastapi_startkit/tests/masoniteorm/fixtures/model.py b/fastapi_startkit/tests/masoniteorm/fixtures/model.py
index 281f3b77..ac761cc9 100644
--- a/fastapi_startkit/tests/masoniteorm/fixtures/model.py
+++ b/fastapi_startkit/tests/masoniteorm/fixtures/model.py
@@ -1,5 +1,8 @@
+from datetime import datetime, timedelta, time, date
+
 from fastapi_startkit.carbon.carbon import Carbon
-from fastapi_startkit.masoniteorm.models.fields import Field, DateTimeField
+from tests.masoniteorm.fixtures.casts import Address
+from fastapi_startkit.masoniteorm import ModelField, Field
 from fastapi_startkit.masoniteorm.relationships import (
     HasOne,
     BelongsTo,
@@ -9,15 +12,20 @@
     HasOneThrough,
     MorphTo,
 )
-from fastapi_startkit.masoniteorm.models.model import Model
+from fastapi_startkit.masoniteorm import Model
 
 
 class User(Model):
     id: int
     name: str
     email: str
-    email_verified_at: Carbon = DateTimeField(fmt="%Y-%m-%d %H:%M:%S", tz="UTC")
+    email_verified_at: datetime
+    date_of_birth: date
+    session_duration: timedelta
+    punch_in_time: time = Field(default=time(12, 0, 0))
     is_admin: bool
+    preferences: dict
+    address: Address = ModelField()
 
     profile: "Profile" = HasOne("Profile", "user_id", "id")
     articles: "Articles" = HasMany("Articles", "id", "user_id")
diff --git a/fastapi_startkit/tests/masoniteorm/fixtures/seeder.py b/fastapi_startkit/tests/masoniteorm/fixtures/seeder.py
index 1f24a0e5..e776fdeb 100644
--- a/fastapi_startkit/tests/masoniteorm/fixtures/seeder.py
+++ b/fastapi_startkit/tests/masoniteorm/fixtures/seeder.py
@@ -1,9 +1,19 @@
-from .model import User, Profile, Articles, Logo, Country, Port, IncomingShipment, Like, Product
+from .model import Articles, Country, IncomingShipment, Like, Logo, Port, Product, Profile, User
 
 
 async def seeder():
     user = await User.query().create(
-        {"email": "admin@admin.com", "name": "Joe", "is_admin": True}
+        {
+            "email": "admin@admin.com",
+            "name": "Joe",
+            "is_admin": True,
+            "email_verified_at": "2024-01-15 08:00:00",
+            "date_of_birth": "1990-06-15",
+            "session_duration": 3600.0,
+            "punch_in_time": "09:00:00",
+            "preferences": {"theme": "dark", "language": "en"},
+            "address": {"address": "123 Main St", "city": "Sydney", "state": "NSW", "country": "Australia"},
+        }
     )
     await Profile.create({"name": "Joe Profile", "user_id": user.id})
     article = await Articles.create(
@@ -13,9 +23,7 @@ async def seeder():
             "published_date": "2020-01-01 00:00:00",
         }
     )
-    await Logo.create(
-        {"article_id": article.id, "published_date": "2020-01-01 00:00:00"}
-    )
+    await Logo.create({"article_id": article.id, "published_date": "2020-01-01 00:00:00"})
     product = await Product.create({"name": "Widget"})
     await Like.create({"likeable_type": "article", "likeable_id": article.id})
     await Like.create({"likeable_type": "product", "likeable_id": product.id})
diff --git a/fastapi_startkit/tests/masoniteorm/sqlite/models/test_sqlite_model_casts.py b/fastapi_startkit/tests/masoniteorm/sqlite/models/test_sqlite_model_casts.py
new file mode 100644
index 00000000..d2bcaa29
--- /dev/null
+++ b/fastapi_startkit/tests/masoniteorm/sqlite/models/test_sqlite_model_casts.py
@@ -0,0 +1,166 @@
+import datetime
+
+import pendulum
+
+from ...fixtures.casts import Address
+from ...fixtures.model import User
+from ..test_case import TestCase
+
+
+class SqliteModelCastsTest(TestCase):
+    async def test_sqlite_model_casts_int(self):
+        user = await User.first()
+
+        # id: int annotation → IntCast
+        self.assertIsInstance(user.id, int)
+
+    async def test_sqlite_model_casts_str(self):
+        user = await User.first()
+
+        # name: str annotation → str cast
+        self.assertIsInstance(user.name, str)
+
+    async def test_sqlite_model_casts_bool_true(self):
+        user = await User.first()
+
+        # is_admin: bool annotation → BoolCast; SQLite stores booleans as 0/1
+        self.assertIsInstance(user.is_admin, bool)
+        self.assertTrue(user.is_admin)
+
+    async def test_sqlite_model_casts_bool_false(self):
+        user = await User.where("email", "guest@guest.com").first()
+
+        # A user seeded with is_admin=False should cast to bool False
+        self.assertIsInstance(user.is_admin, bool)
+        self.assertFalse(user.is_admin)
+
+    async def test_sqlite_model_casts_dict(self):
+        user = await User.where("email", "admin@admin.com").first()
+
+        # preferences: dict annotation → JsonCast; stored as JSON string in SQLite
+        self.assertIsInstance(user.preferences, dict)
+        self.assertEqual(user.preferences["theme"], "dark")
+        self.assertEqual(user.preferences["language"], "en")
+
+    async def test_sqlite_model_casts_dict_none(self):
+        user = await User.where("email", "guest@guest.com").first()
+
+        # guest user has no preferences seeded — should remain None
+        self.assertIsNone(user.preferences)
+
+    async def test_sqlite_model_casts_list(self):
+        user = await User.where("email", "admin@admin.com").first()
+
+        # Update preferences to a JSON array and verify list cast on re-fetch
+        await user.update({"preferences": ["reading", "coding"]})
+        updated = await User.where("email", "admin@admin.com").first()
+
+        self.assertIsInstance(updated.preferences, list)
+        self.assertIn("reading", updated.preferences)
+        self.assertIn("coding", updated.preferences)
+
+    async def test_sqlite_model_casts_pydantic_object(self):
+        user = await User.where("email", "admin@admin.com").first()
+
+        # address: Address annotation → custom Pydantic cast
+        # DB stores JSON text, get() deserializes it into an Address instance
+        self.assertIsInstance(user.address, Address)
+        self.assertEqual(user.address.address, "123 Main St")
+        self.assertEqual(user.address.city, "Sydney")
+        self.assertEqual(user.address.state, "NSW")
+        self.assertEqual(user.address.country, "Australia")
+
+    async def test_sqlite_model_casts_pydantic_object_none(self):
+        user = await User.where("email", "guest@guest.com").first()
+
+        # guest user has no address seeded — should remain None
+        self.assertIsNone(user.address)
+
+    async def test_sqlite_model_casts_pydantic_object_insert_with_instance(self):
+        address = Address(address="456 Queen St", city="Melbourne", state="VIC", country="Australia")
+
+        await User.create({
+            "email": "instance@example.com",
+            "name": "Instance User",
+            "is_admin": False,
+            "address": address,
+        })
+
+        fetched = await User.where("email", "instance@example.com").first()
+
+        self.assertIsInstance(fetched.address, Address)
+        self.assertEqual(fetched.address.address, "456 Queen St")
+        self.assertEqual(fetched.address.city, "Melbourne")
+
+    async def test_sqlite_model_casts_pydantic_object_insert_with_dict(self):
+        await User.create({
+            "email": "dict@example.com",
+            "name": "Dict User",
+            "is_admin": False,
+            "address": {"address": "789 King St", "city": "Brisbane", "state": "QLD", "country": "Australia"},
+        })
+
+        fetched = await User.where("email", "dict@example.com").first()
+
+        self.assertIsInstance(fetched.address, Address)
+        self.assertEqual(fetched.address.address, "789 King St")
+        self.assertEqual(fetched.address.city, "Brisbane")
+
+    async def test_sqlite_model_casts_datetime(self):
+        await User.create({
+            "email": "datetime@example.com",
+            "name": "DateTime User",
+            "is_admin": False,
+            "email_verified_at": "2024-06-15 12:30:00",
+            "date_of_birth": datetime.datetime.now(datetime.timezone.utc),
+        })
+
+        user = await User.where("email", "datetime@example.com").first()
+
+        # email_verified_at: Carbon = DateTimeField() → DateCast → pendulum.DateTime
+        self.assertIsInstance(user.email_verified_at, pendulum.DateTime)
+        self.assertEqual(user.email_verified_at.year, 2024)
+        self.assertEqual(user.email_verified_at.month, 6)
+        self.assertEqual(user.email_verified_at.day, 15)
+
+    async def test_sqlite_model_casts_datetime_none(self):
+        user = await User.where("email", "guest@guest.com").first()
+
+        # guest user has no email_verified_at seeded — should remain None
+        self.assertIsNone(user.email_verified_at)
+
+    async def test_sqlite_model_casts_date(self):
+        user = await User.where("email", "admin@admin.com").first()
+
+        # date_of_birth: date annotation → DateCast → pendulum.DateTime
+        self.assertIsInstance(user.date_of_birth, pendulum.DateTime)
+        self.assertEqual(user.date_of_birth.year, 1990)
+        self.assertEqual(user.date_of_birth.month, 6)
+        self.assertEqual(user.date_of_birth.day, 15)
+
+    async def test_sqlite_model_casts_timedelta(self):
+        user = await User.where("email", "admin@admin.com").first()
+
+        # session_duration: timedelta annotation → TimeDeltaCast; stored as seconds
+        self.assertIsInstance(user.session_duration, datetime.timedelta)
+        self.assertEqual(user.session_duration.total_seconds(), 3600.0)
+
+    async def test_sqlite_model_casts_time(self):
+        user = await User.where("email", "admin@admin.com").first()
+
+        # punch_in_time: time annotation → TimeCast; stored as HH:MM:SS string
+        self.assertIsInstance(user.punch_in_time, datetime.time)
+        self.assertEqual(user.punch_in_time.hour, 9)
+        self.assertEqual(user.punch_in_time.minute, 0)
+
+    async def test_sqlite_model_casts_timedelta_none(self):
+        user = await User.where("email", "guest@guest.com").first()
+
+        self.assertIsNone(user.session_duration)
+
+    async def test_sqlite_model_casts_time_default(self):
+        user = await User.where("email", "guest@guest.com").first()
+
+        # guest has no punch_in_time seeded (NULL) — Field(default=time(12, 0, 0)) should apply
+        self.assertIsInstance(user.punch_in_time, datetime.time)
+        self.assertEqual(user.punch_in_time, datetime.time(12, 0, 0))
diff --git a/fastapi_startkit/uv.lock b/fastapi_startkit/uv.lock
index f1ae20be..dd3569e8 100644
--- a/fastapi_startkit/uv.lock
+++ b/fastapi_startkit/uv.lock
@@ -443,7 +443,7 @@ wheels = [
 
 [[package]]
 name = "fastapi-startkit"
-version = "0.24.0"
+version = "0.25.0"
 source = { editable = "." }
 dependencies = [
     { name = "cleo" },
diff --git a/queue.py b/queue.py
new file mode 100644
index 00000000..52902438
--- /dev/null
+++ b/queue.py
@@ -0,0 +1,346 @@
+from __future__ import annotations
+
+import asyncio
+import importlib
+import inspect
+import json
+import time
+import uuid
+from abc import ABC, abstractmethod
+from collections import deque
+from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
+from datetime import datetime, timezone
+from typing import Any
+
+
+# ---------------------------------------------------------------------------
+# Pool strategies
+# ---------------------------------------------------------------------------
+
+class Pool(ABC):
+    """Abstract base class that all pool strategies must implement."""
+
+    @abstractmethod
+    def submit(self, payload: dict) -> None:
+        """Dispatch a serialised job payload for execution."""
+
+    def shutdown(self) -> None:
+        """Clean up resources. Override when the pool holds an executor."""
+
+    @staticmethod
+    def parse(payload: dict) -> tuple[str, str]:
+        """Return (fully-qualified class path, method name) from a payload."""
+        return payload["class"], payload["method"]
+
+    @staticmethod
+    def resolve_and_call(payload: dict) -> None:
+        """Deserialise a payload, instantiate the job class, and call its method."""
+        class_path, method = Pool.parse(payload)
+        module_name, class_name = class_path.rsplit(".", 1)
+        cls = getattr(importlib.import_module(module_name), class_name)
+        result = getattr(cls(**payload.get("data", {})), method)()
+        if asyncio.iscoroutine(result):
+            asyncio.run(result)
+
+
+class ThreadPool(Pool):
+    """Each job runs in a worker thread; async jobs get their own event loop."""
+
+    def __init__(self, max_workers: int = 4) -> None:
+        self._executor = ThreadPoolExecutor(max_workers=max_workers)
+
+    def submit(self, payload: dict) -> None:
+        self._executor.submit(Pool.resolve_and_call, payload)
+
+    def shutdown(self) -> None:
+        self._executor.shutdown(wait=True)
+
+
+class AsyncPool(Pool):
+    """Jobs run on the current event loop as asyncio Tasks."""
+
+    def __init__(self) -> None:
+        self._tasks: list[asyncio.Task] = []
+
+    def submit(self, payload: dict) -> None:
+        loop = asyncio.get_event_loop()
+        task = loop.create_task(self._run(payload))
+        self._tasks.append(task)
+
+    @staticmethod
+    async def _run(payload: dict) -> None:
+        class_path, method = Pool.parse(payload)
+        module_name, class_name = class_path.rsplit(".", 1)
+        cls = getattr(importlib.import_module(module_name), class_name)
+        result = getattr(cls(**payload.get("data", {})), method)()
+        if asyncio.iscoroutine(result):
+            await result
+
+    async def drain(self) -> None:
+        if self._tasks:
+            await asyncio.gather(*self._tasks, return_exceptions=True)
+            self._tasks.clear()
+
+
+class ProcessPool(Pool):
+    """Each job runs in a separate OS process via asyncio.run() internally."""
+
+    def __init__(self, max_workers: int = 4) -> None:
+        self._executor = ProcessPoolExecutor(max_workers=max_workers)
+
+    def submit(self, payload: dict) -> None:
+        # Pool.resolve_and_call is a static method on an importable class — picklable
+        self._executor.submit(Pool.resolve_and_call, payload)
+
+    def shutdown(self) -> None:
+        self._executor.shutdown(wait=True)
+
+
+# ---------------------------------------------------------------------------
+# Queue
+# ---------------------------------------------------------------------------
+
+class Queue:
+    def __init__(self) -> None:
+        self._store: deque[dict] = deque()
+        self.failed: list[dict] = []
+
+    def push(self, job: "Job") -> None:
+        self._store.append(job.payload())
+
+    def pop(self) -> dict | None:
+        return self._store.popleft() if self._store else None
+
+    def size(self) -> int:
+        return len(self._store)
+
+    @staticmethod
+    def fire(payload: dict) -> None:
+        Pool.resolve_and_call(payload)
+
+class DatabaseQueue(Queue):
+    """Database-backed queue. Persists jobs to a `jobs` table via SQLAlchemy."""
+
+    def __init__(self, connection_name: str = "default") -> None:
+        super().__init__()
+        self._connection_name = connection_name
+        self._db: Any = None   # inject an AsyncSession / connection here
+
+    # -- public ---------------------------------------------------------------
+
+    def push(self, job: "Job", queue: str = "default", delay: int = 0) -> str:
+        payload = self.create_payload(job, queue)
+        return self.push_to_database(queue, payload, delay, attempts=0)
+
+    def later(self, job: "Job", delay: int, queue: str = "default") -> str:
+        """Push a job with a delay (seconds from now)."""
+        return self.push(job, queue=queue, delay=delay)
+
+    # -- payload --------------------------------------------------------------
+
+    def create_payload(self, job: "Job", queue: str = "default") -> dict:
+        class_path = f"{job.__class__.__module__}.{job.__class__.__name__}"
+        return {
+            "uuid":          str(uuid.uuid4()),
+            "display_name":  self._get_display_name(job),
+            "job":           class_path,
+            "max_tries":     getattr(job, "tries", None),
+            "max_exceptions":getattr(job, "max_exceptions", None),
+            "fail_on_timeout":getattr(job, "fail_on_timeout", False),
+            "backoff":       self._get_backoff(job),
+            "timeout":       getattr(job, "timeout", None),
+            "retry_until":   self._get_expiration(job),
+            "data": {
+                "command_name": class_path,
+                "command":      job.payload(),
+                "batch_id":     getattr(job, "batch_id", None),
+            },
+            "created_at":    int(datetime.now(timezone.utc).timestamp()),
+        }
+
+    # -- database interaction -------------------------------------------------
+
+    def push_to_database(
+        self, queue: str, payload: dict, delay: int, attempts: int
+    ) -> str:
+        available_at = int(datetime.now(timezone.utc).timestamp()) + delay
+
+        # Example row — replace with: await JobModel.create(...)
+        row = {
+            "uuid":         payload["uuid"],
+            "queue":        queue,
+            "payload":      payload,
+            "attempts":     attempts,
+            "available_at": available_at,
+            "created_at":   payload["created_at"],
+        }
+        _ = row   # placeholder until DB model is wired in
+        return payload["uuid"]
+
+    def pop(self) -> dict | None:
+        """
+        Fetch the next available job from the database.
+
+        Override with: SELECT ... WHERE available_at <= NOW() ORDER BY id LIMIT 1.
+        Falls back to in-memory store when no DB is wired.
+        """
+        return super().pop()
+
+    # -- helpers --------------------------------------------------------------
+
+    @staticmethod
+    def _get_display_name(job: "Job") -> str:
+        return job.__class__.__name__
+
+    @staticmethod
+    def _get_backoff(job: "Job") -> int | list[int] | None:
+        return getattr(job, "backoff", None)
+
+    @staticmethod
+    def _get_expiration(job: "Job") -> int | None:
+        retry_until = getattr(job, "retry_until", None)
+        if callable(retry_until):
+            return int(retry_until().timestamp())
+        return int(retry_until.timestamp()) if retry_until else None
+
+class RedisQueue(Queue):
+    """Redis-backed queue (lpush / brpop)."""
+
+    def __init__(self, connection_name: str = "default") -> None:
+        super().__init__()
+        self._connection_name = connection_name
+        self._client: Any = None   # inject a redis.Redis / aioredis client here
+
+    def push(self, job: "Job") -> None:
+        if self._client:
+            import json
+            self._client.lpush(self._connection_name, json.dumps(job.payload()))
+        else:
+            super().push(job)   # fall back to in-memory while not connected
+
+    def pop(self) -> dict | None:
+        if self._client:
+            import json
+            result = self._client.brpop(self._connection_name, timeout=1)
+            return json.loads(result[1]) if result else None
+        return super().pop()
+
+
+# ---------------------------------------------------------------------------
+# Worker
+# ---------------------------------------------------------------------------
+
+class Worker:
+    def __init__(
+        self,
+        queue: Queue,
+        pool: Pool | None = None,
+        max_jobs: int | None = None,
+    ) -> None:
+        self._queue = queue
+        self._pool = pool or ThreadPool()
+        self._max_jobs = max_jobs   # None = run forever (real daemon)
+
+    def daemon(self) -> None:
+        start_time, job_processed = time.time(), 0
+
+        while True:
+            job = self._queue.pop()
+
+            if job is not None:                  # fixed: was `if job None:`
+                job_processed += 1
+                self.run_job(job)
+
+            if self._max_jobs and job_processed >= self._max_jobs:
+                break
+
+            if job is None:
+                time.sleep(0.05)
+
+        self._pool.shutdown()
+
+        elapsed = time.time() - start_time
+        print(f"[worker] processed {job_processed} job(s) in {elapsed:.3f}s")
+
+    def run_job(self, job: dict) -> None:        # fixed: removed unused `connection_name` param
+        try:
+            self.process(job)
+        except Exception as e:
+            self.report(e, job)
+            self.stop_worker_if_connection_lost(e)
+
+    def process(self, job: dict) -> None:
+        self._pool.submit(job)                   # fixed: exceptions now propagate out of process()
+
+    def report(self, e: Exception, job: dict | None = None) -> None:
+        label = job.get("class", "unknown") if job else "unknown"
+        print(f"[worker] job failed — {label}: {e}")
+        if job:
+            self._queue.failed.append({"payload": job, "error": str(e)})
+
+    @staticmethod
+    def stop_worker_if_connection_lost(e: Exception) -> None:
+        connection_errors = (ConnectionError, OSError)
+        if isinstance(e, connection_errors):
+            raise SystemExit(f"[worker] stopping — connection lost: {e}")
+
+
+# ---------------------------------------------------------------------------
+# Base Job
+# ---------------------------------------------------------------------------
+
+class Job:
+    def payload(self) -> dict:
+        return {
+            "class": f"{self.__class__.__module__}.{self.__class__.__name__}",
+            "method": "handle",
+            "data": self._data(),
+        }
+
+    def _data(self) -> dict:
+        """Serialise all constructor arguments automatically from instance attributes."""
+        params = inspect.signature(self.__class__.__init__).parameters
+        return {
+            name: getattr(self, name)
+            for name in params
+            if name != "self" and hasattr(self, name)
+        }
+
+    def handle(self) -> None:
+        raise NotImplementedError
+
+
+# ---------------------------------------------------------------------------
+# EmailQueue  (specialised queue for mail jobs)
+# ---------------------------------------------------------------------------
+
+class EmailQueue(Queue):
+    async def handle(self) -> None:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Demo  (python queue.py)
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    # -- define a quick inline job for the demo ------------------------------
+    class WelcomeEmail(Job):
+        def __init__(self, email: str, name: str = "there") -> None:
+            self.email = email
+            self.name = name
+
+        def handle(self) -> None:
+            print(f"  [job] WelcomeEmail → to={self.email}, name={self.name}")
+
+    # -------------------------------------------------------------------------
+
+    q = Queue()
+    q.push(WelcomeEmail(email="alice@example.com", name="Alice"))
+    q.push(WelcomeEmail(email="bob@example.com"))
+
+    print(f"[queue] {q.size()} job(s) queued\n")
+
+    # uv run artisan queue:work --pool=thread (default)
+    worker = Worker(q, pool=ThreadPool(max_workers=2), max_jobs=2)
+    worker.daemon()