fix: engine bugs and python sdk (#178)

* fix: engine bugs and python sdk

* chore: fix linting

internal/schema/schema.go

@@ -95,13 +95,17 @@ func parseArray(arr []interface{}) reflect.Type {
 
 var nameRegex = regexp.MustCompile(`(\b|-|_|\.)[a-z]`)
 
+var invalidCharRegex = regexp.MustCompile(`[^a-zA-Z0-9_]`)
+
 // toExportedName converts a JSON key into an exported Go field name.
 func toExportedName(key string) string {
-	return nameRegex.ReplaceAllStringFunc(key, func(t string) string {
+	res := nameRegex.ReplaceAllStringFunc(key, func(t string) string {
 		if len(t) == 1 {
 			return strings.ToUpper(t)
 		}
 
 		return strings.ToUpper(string(t[1]))
 	})
+
+	return invalidCharRegex.ReplaceAllString(res, "")
 }

internal/services/grpc/server.go

@@ -179,7 +179,21 @@ func (s *Server) startGRPC(ctx context.Context) error {
 	authMiddleware := middleware.NewAuthN(s.config)
 
 	grpcPanicRecoveryHandler := func(p any) (err error) {
-		s.l.Err(p.(error)).Msgf("recovered from panic: %s", string(debug.Stack()))
+		panicErr, ok := p.(error)
+
+		var panicStr string
+
+		if !ok {
+			panicStr, ok = p.(string)
+
+			if !ok {
+				panicStr = "Could not determine panic error"
+			}
+		} else {
+			panicStr = panicErr.Error()
+		}
+
+		s.l.Error().Msgf("recovered from panic: %s. Stack: %s", panicStr, string(debug.Stack()))
 		return status.Errorf(codes.Internal, "An internal error occurred")
 	}
 

internal/services/worker/worker.go

@@ -12,9 +12,12 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
+	"github.com/rs/zerolog"
+
 	"github.com/hatchet-dev/hatchet/internal/encryption"
 	"github.com/hatchet-dev/hatchet/internal/integrations/vcs"
 	"github.com/hatchet-dev/hatchet/internal/integrations/vcs/vcsutils"
+	"github.com/hatchet-dev/hatchet/internal/logger"
 	"github.com/hatchet-dev/hatchet/internal/repository"
 	"github.com/hatchet-dev/hatchet/pkg/client"
 	"github.com/hatchet-dev/hatchet/pkg/worker"
@@ -36,12 +39,17 @@ type WorkerOpt func(*WorkerOpts)
 type WorkerOpts struct {
 	client client.Client
 
+	l            *zerolog.Logger
 	repo         repository.Repository
 	vcsProviders map[vcs.VCSRepositoryKind]vcs.VCSProvider
 }
 
 func defaultWorkerOpts() *WorkerOpts {
-	return &WorkerOpts{}
+	logger := logger.NewDefaultLogger("internal-worker")
+
+	return &WorkerOpts{
+		l: &logger,
+	}
 }
 
 func WithRepository(r repository.Repository) WorkerOpt {
@@ -62,9 +70,16 @@ func WithClient(c client.Client) WorkerOpt {
 	}
 }
 
+func WithLogger(l *zerolog.Logger) WorkerOpt {
+	return func(opts *WorkerOpts) {
+		opts.l = l
+	}
+}
+
 type WorkerImpl struct {
 	*worker.Worker
 
+	l            *zerolog.Logger
 	repo         repository.Repository
 	vcsProviders map[vcs.VCSRepositoryKind]vcs.VCSProvider
 }
@@ -96,6 +111,7 @@ func NewWorker(fs ...WorkerOpt) (*WorkerImpl, error) {
 		Worker:       hatchetWorker,
 		repo:         opts.repo,
 		vcsProviders: opts.vcsProviders,
+		l:            opts.l,
 	}, nil
 }
 

python-sdk/examples/dag/worker.py

@@ -12,11 +12,11 @@ class MyWorkflow:
 
     @hatchet.step()
     def step1(self, context : Context):
-        context.overrides("test", "test")
+        overrideValue = context.overrides("prompt", "You are an AI assistant...")
 
         print("executed step1", context.workflow_input())
         return {
-            "step1": "step1",
+            "step1": overrideValue,
         }
 
     @hatchet.step()

python-sdk/hatchet_sdk/clients/listener.py

@@ -87,8 +87,12 @@ class HatchetListener:
                             raise Exception(
                                 f"Unknown event type: {workflow_event.eventType}")
                         payload = None
-                        if workflow_event.eventPayload:
-                            payload = json.loads(workflow_event.eventPayload)
+
+                        try:
+                            if workflow_event.eventPayload:
+                                payload = json.loads(workflow_event.eventPayload)
+                        except Exception as e:
+                            pass
 
                         # call the handler
                         event = StepRunEvent(type=eventType, payload=payload)
@@ -101,8 +105,12 @@ class HatchetListener:
                                 f"Unknown event type: {workflow_event.eventType}")
                         
                         payload = None
-                        if workflow_event.eventPayload:
-                            payload = json.loads(workflow_event.eventPayload)
+                        
+                        try:
+                            if workflow_event.eventPayload:
+                                payload = json.loads(workflow_event.eventPayload)
+                        except Exception as e:
+                            pass
                         
                     if workflow_event.hangup:
                         listener = None

python-sdk/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "hatchet-sdk"
-version = "0.10.0"
+version = "0.10.1"
 description = ""
 authors = ["Alexander Belanger <alexander@hatchet.run>"]
 readme = "README.md"