Merge pull request #338 from MasoniteFramework/fix/334

Fix/334

Author: josephmancuso

app/http/controllers/WelcomeController.py

@@ -2,12 +2,14 @@
 
 from src.masonite.view import View
 from src.masonite.controllers import Controller
+from app.jobs.TestJob import TestJob
+from src.masonite import Queue
 
 
 class WelcomeController(Controller):
     """Controller For Welcoming The User."""
 
-    def show(self, view: View):
+    def show(self, view: View, queue: Queue):
         """Show the welcome page.
 
         Arguments:
@@ -17,4 +19,5 @@ def show(self, view: View):
         Returns:
             masonite.view.View -- The Masonite view class.
         """
+        queue.push(TestJob)
         return view.render('welcome')

craft

@@ -0,0 +1,21 @@
+#!/usr/bin/env python
+"""Craft Command.
+
+This module is really used for backup only if the masonite CLI cannot import this for you.
+This can be used by running "python craft". This module is not ran when the CLI can
+successfully import commands for you.
+"""
+
+from cleo import Application
+from masonite import __version__
+
+from wsgi import container
+
+application = Application('Masonite Version:', __version__)
+
+for key, value in container.providers.items():
+    if isinstance(key, str) and key.endswith('Command'):
+        application.add(container.make('{0}'.format(key)))
+
+if __name__ == '__main__':
+    application.run()

databases/migrations/2020_09_18_233149_create_queue_jobs_table.py

@@ -0,0 +1,19 @@
+from orator.migrations import Migration
+
+
+class CreateQueueJobsTable(Migration):
+    def up(self):
+        """Run the migrations."""
+        with self.schema.create("queue_jobs") as table:
+            table.increments("id")
+            table.string("name")
+            table.binary("serialized")
+            table.integer("attempts")
+            table.integer("failed").nullable()
+            table.timestamp("ran_at").nullable()
+            table.timestamp("created_at").nullable()
+            table.timestamp("wait_until").nullable()
+
+    def down(self):
+        """Revert the migrations."""
+        self.schema.drop("queue_jobs")

routes/web.py

@@ -6,6 +6,7 @@
 ROUTES = [
     Get().route('/test', None).middleware('auth'),
     Get('/bad', 'TestController@bad'),
+    Get('/welcome', 'WelcomeController@show'),
     Get('/keyerror', 'TestController@keyerror'),
     Get().route('/queue', 'TestController@queue'),
     Options('options', 'TestController@show'),

src/masonite/__version__.py

@@ -2,7 +2,7 @@
 __title__ = "masonite"
 __description__ = "The core for the Masonite framework"
 __url__ = "https://github.com/MasoniteFramework/masonite"
-__version__ = "2.3.17"
+__version__ = "2.3.18"
 __author__ = "Joseph Mancuso"
 __author_email__ = "[email protected]"
 __licence__ = "MIT"

src/masonite/commands/SeedCommand.py

@@ -16,7 +16,7 @@ class SeedCommand(Command):
     def handle(self):
         table = self.argument("table").lower()
         subprocess.call(
-            ["orator make:seed {}_table_seeder -p databases/seeds".format(table),],
+            ["orator make:seed {}_table_seeder -p databases/seeds".format(table)],
             shell=True,
         )
 

src/masonite/drivers/queue/QueueDatabaseDriver.py

@@ -75,30 +75,24 @@ def consume(self, channel, **options):  # skipcq
         )
         schema = schema.connection(channel)
         while True:
-            jobs = schema.table("queue_jobs").where("ran_at", None).get()
+            builder = schema.table("queue_jobs")
+            jobs = builder.where_null("ran_at").where(schema.table("queue_jobs").where_null('wait_until').or_where('wait_until', '<=', pendulum.now().to_datetime_string())).limit(1).get()
+
             if not jobs.count():
                 time.sleep(5)
 
             for job in jobs:
+                builder.where("id", job["id"]).update(
+                    {
+                        "ran_at": pendulum.now().to_datetime_string(),
+                    }
+                )
                 unserialized = pickle.loads(job.serialized)
                 obj = unserialized["obj"]
                 args = unserialized["args"]
                 callback = unserialized["callback"]
                 ran = job.attempts
 
-                wait_time = job["wait_until"]
-
-                if not job["wait_until"]:
-                    wait_time = pendulum.now()
-                else:
-                    if isinstance(wait_time, str):
-                        wait_time = pendulum.parse(job["wait_until"])
-                    else:
-                        wait_time = pendulum.instance(job["wait_until"])
-
-                # print(job['wait_until'], wait_time.is_future())
-                if job["wait_until"] and wait_time.is_future():
-                    continue
                 try:
                     try:
                         if inspect.isclass(obj):
@@ -111,7 +105,7 @@ def consume(self, channel, **options):  # skipcq
 
                     try:
                         # attempts = 1
-                        schema.table("queue_jobs").where("id", job["id"]).update(
+                        builder.where("id", job["id"]).update(
                             {
                                 "ran_at": pendulum.now().to_datetime_string(),
                                 "attempts": job["attempts"] + 1,
@@ -125,7 +119,7 @@ def consume(self, channel, **options):  # skipcq
 
                     if not obj.run_again_on_fail:
                         # ch.basic_ack(delivery_tag=method.delivery_tag)
-                        schema.table("queue_jobs").where("id", job["id"]).update(
+                        builder.where("id", job["id"]).update(
                             {
                                 "ran_at": pendulum.now().to_datetime_string(),
                                 "failed": 1,
@@ -135,12 +129,12 @@ def consume(self, channel, **options):  # skipcq
 
                     if ran < obj.run_times and isinstance(obj, Queueable):
                         time.sleep(1)
-                        schema.table("queue_jobs").where("id", job["id"]).update(
-                            {"attempts": job["attempts"] + 1,}
+                        builder.where("id", job["id"]).update(
+                            {"attempts": job["attempts"] + 1}
                         )
                         continue
                     else:
-                        schema.table("queue_jobs").where("id", job["id"]).update(
+                        builder.where("id", job["id"]).update(
                             {
                                 "attempts": job["attempts"] + 1,
                                 "ran_at": pendulum.now().to_datetime_string(),

src/masonite/providers/UploadProvider.py

@@ -24,5 +24,5 @@ def boot(self, manager: UploadManager, view: View):
         self.app.bind("Upload", manager.driver(config("storage").DRIVER))
         self.app.swap(Upload, manager.driver(config("storage").DRIVER))
         view.share(
-            {"static": static,}
+            {"static": static}
         )

src/masonite/testing/TestCase.py

@@ -141,7 +141,7 @@ def call(self, method, url, params, wsgi={}):
             )
 
         custom_wsgi.update(
-            {"QUERY_STRING": urlencode(params),}
+            {"QUERY_STRING": urlencode(params)}
         )
 
         self.run_container(custom_wsgi)