Celery with SQS

I was reading a blog post about SQS that was not working well with Airflow .
https://medium.com/@Twistacz/whats-wrong-with-sqs-and-why-we-gave-it-up-22780d4f48ef
Is this the case now ???
If so what is the best broker to make it work in AWS environment .

I didn't observe any issues with SQS as a backend. There are observability downsides and upsides. There's no way to use Celery Flower, for example, but CloudWatch alarms are very more versatile.

This is not related to celery !!! There is 1 issue I face regularly when I setup python environment . In requirements.txt if I have the version airflow 1.9.0 next time in y deployment if I increment my version to 1.10.0 then according to the script it directly installs 1.10.0 which causes issues because dependencies of 1.10.0 will be different . I way I thought it can be handles was have a Virtual env which installs and remove the env every time it is deployed and recreate the env and run it in the new env . Let me know if I am thinking in the right direction.


Feature request: constructor injection

Thanks for this really nice implementation of MessagePack!
We use a lot of immutable objects in our code, and these are currently not very compatible with this implementation. It would be nice if the serializer could detect the possibility of deserializing an object via the constructor if there are no settable properties. Current, this implementation throws the following exception in this scenario:
Cannot serialize type 'ConstructorInjectionDemo' because it does not have any serializable fields nor properties
I've written a sample test that demonstrates the feature I'm proposing here:
using System.IO;
using MsgPack.Serialization;
using NUnit.Framework;

public class ConstructorInjectionDemo
{
    public string Value { get; private set; }

    public ConstructorInjectionDemo(string value)
    {
        Value = value;
    }
}

[TestFixture]
public class MessagePackConstructorInjectionTests
{
    [Test]
    public void Given_a_class_without_public_setters_When_MessagePack_deserializes_the_object_Then_the_constructor_should_be_used()
    {
        var serializer = MessagePackSerializer.Get<ConstructorInjectionDemo>();
        using(var stream = new MemoryStream())
        {
            serializer.Pack(stream, new ConstructorInjectionDemo("Hello!"));
            stream.Position = 0;
            var deserialisedObject = serializer.Unpack(stream);
            Assert.That(deserialisedObject.Value, Is.EqualTo("Hello!"));
        }
    }
}

Is there any appetite to implement this feature?
Thanks,
Bart

Any news on this issue? I'm facing this exact issue - immutable types initialized via constructor injection. I've had to break immutability on my models and I'm not comfortable with that.

@jakesays
Sorry, long time passed since last answer. I've just pushed local work and put pre-release version (0.6.0-Alpha1) on NuGet. I'm happy if you try it and post feedback :)

@yfakariya Excellent. Thanks! i'll give it a try now.
Which method did you choose? After my message yesterday I implemented ctor based deserialization using matching ctor parameter names/property names.

@yfakariya,
I gave your changes a try and came up with an issue - the unpacker is throwing FatalExecutionEngineError.
First, here's my test app:

using MsgPack.Serialization;

namespace MsgPackTester
{
	public struct ObjectId
	{
		public long Number { get; private set; }
		public int Version { get; private set; }

		public ObjectId(long number, int version)
			: this()
		{
			Number = number;
			Version = version;
		}
	}

	class Program
	{
		static void Main(string[] args)
		{
			var serializer = MessagePackSerializer.Get<ObjectId>();

			var id = new ObjectId(10, 1);

			//works. produces a 3 byte array.
			var bits = serializer.PackSingleObject(id);

			//crashes with FatalExecutionEngineError
			var id2 = serializer.UnpackSingleObject(bits);
		}
	}
}

And here is the decompiled serializer for ObjectId:

using MsgPack;
using MsgPack.Serialization;
using MsgPackTester;
using System;
using System.Reflection;

namespace MsgPack.Serialization.EmittingSerializers.Generated
{
    public sealed class MsgPackTester_ObjectIdSerializer : MessagePackSerializer<ObjectId>
    {
        private readonly MessagePackSerializer<long> _serializer0;

        private readonly MessagePackSerializer<int> _serializer1;

        private readonly FieldInfo _functionObjectId_set_Number0;

        private readonly FieldInfo _functionObjectId_set_Version1;

        public MsgPackTester_ObjectIdSerializer() : this(null)
        {
        }

        public MsgPackTester_ObjectIdSerializer(SerializationContext serializationContext) : base(serializationContext)
        {
            this._serializer0 = serializationContext.GetSerializer<long>();
            this._serializer1 = serializationContext.GetSerializer<int>();
            this._functionObjectId_set_Number0 = MethodBase.GetMethodFromHandle(typeof(ObjectId).GetMethod("set_Number", new Type[] { typeof(long) }).MethodHandle);
            this._functionObjectId_set_Version1 = MethodBase.GetMethodFromHandle(typeof(ObjectId).GetMethod("set_Version", new Type[] { typeof(int) }).MethodHandle);
        }

        protected sealed override void PackToCore(Packer packer, ObjectId objectId)
        {
            packer.PackArrayHeader(2);
            this._serializer0.PackTo(packer, objectId.Number);
            this._serializer1.PackTo(packer, objectId.Version);
        }

        protected sealed override ObjectId UnpackFromCore(Unpacker unpackers)
        {
            ObjectId objectId = new ObjectId();
            int num = 0;
            long? nullable = new long?();
            int? nullable1 = new int?();
            int num1 = 0;
            if (!unpackers.IsArrayHeader)
            {
                int itemsCount = UnpackHelpers.GetItemsCount(unpackers);
                while (num1 < itemsCount)
                {
                    string str = UnpackHelpers.UnpackStringValue(unpackers, typeof(ObjectId), "MemberName");
                    if (str == null)
                    {
                        throw SerializationExceptions.NewNullIsProhibited("MemberName");
                    }
                    string str1 = str;
                    if (str1 == "Version")
                    {
                        int? nullable2 = UnpackHelpers.UnpackNullableInt32Value(unpackers, typeof(ObjectId), "Int32 Version");
                        if (nullable2.HasValue)
                        {
                            FieldInfo _functionObjectIdSetVersion1 = this._functionObjectId_set_Version1;
                            object[] value = new object[] { nullable2.Value };
                            ((MethodBase)_functionObjectIdSetVersion1).Invoke(objectId, value);
                        }
                    }
                    else if (str1 != "Number")
                    {
                        unpackers.Skip();
                    }
                    else
                    {
                        long? nullable3 = UnpackHelpers.UnpackNullableInt64Value(unpackers, typeof(ObjectId), "Int64 Number");
                        if (nullable3.HasValue)
                        {
                            FieldInfo _functionObjectIdSetNumber0 = this._functionObjectId_set_Number0;
                            object[] objArray = new object[] { nullable3.Value };
                            ((MethodBase)_functionObjectIdSetNumber0).Invoke(objectId, objArray);
                        }
                    }
                    num1++;
                }
            }
            else
            {
                int itemsCount1 = UnpackHelpers.GetItemsCount(unpackers);
                if (num < itemsCount1)
                {
                    nullable = UnpackHelpers.UnpackNullableInt64Value(unpackers, typeof(ObjectId), "Int64 Number");
                }
                if (nullable.HasValue)
                {
                    FieldInfo fieldInfo = this._functionObjectId_set_Number0;
                    object[] value1 = new object[] { nullable.Value };
                    ((MethodBase)fieldInfo).Invoke(objectId, value1);
                }
                num++;
                if (num < itemsCount1)
                {
                    nullable1 = UnpackHelpers.UnpackNullableInt32Value(unpackers, typeof(ObjectId), "Int32 Version");
                }
                if (nullable1.HasValue)
                {
                    FieldInfo _functionObjectIdSetVersion11 = this._functionObjectId_set_Version1;
                    object[] objArray1 = new object[] { nullable1.Value };
                    ((MethodBase)_functionObjectIdSetVersion11).Invoke(objectId, objArray1);
                }
                num++;
            }
            return objectId;
        }
    }
}

I suspect the issue is private readonly FieldInfo _functionObjectId_set_Number0. ObjectId exposes properties only, you're treating them as fields.
I also noticed the unpacker is using reflection to set the members directly. Is this what you expected? I was hoping the ObjectId(long number, int version) would be used to initialize the unpacked instance.
Thanks for your help!
-Jake

I think i found the issue.
If I change SerializationTarget.cs:310 from
if ( asProperty.GetSetMethod( true ) != null )
to
if ( asProperty.GetSetMethod( false ) != null )
then I get the proper constructor being called.
The FieldInfo thing is still going to be an issue, though.
-Jake

Thank you for reporting. I'll add more test for private setter and will fix it in the next release, sorry.

@jakesays I released this feature in 0.6.0-beta2. Please try it. Thanks!

Fixed in 0.6.0.


operator: made the unrecognized argument error clearer

Hey folks,
I was working on something on devito, and got stuck up to an hour trying to figure out that this error:
ValueError: Unrecognized argument damp=damp(x, y).

meant the operator expressions didn't have the function damp. Thinking back it is obvious but the error message could have been cleaerer. I'm just making this commit as way to prevent another person to go through this. If I encounter other situations such as this I will open another PR so that we can make the error messages and error catching mechanisms work better.
I'm not sure if that is the final solution to the problem, but I hope this can at least start a discussion.
Cheers!

Thanks!
I'm not entirely convinced that's the most likely case one could bump into such an exception? Not sure. @mloubout thoughts ?

for example, it could also be a name mismatch
u0 = TimeFunction(name='u', ...)
u1 = TimeFunction(name='u', ...)

eq = Eq(u.forward, u + 1)

op = Operator(eq)

op(u0=u1)

I've seen this happening too

I get the message can be a little bit cryptic now but not sure this update covers all the case or make it more clear. It may be a bit messy to try to cover all the cases as it would need to infer what the user is trying to pass to the Operator.

I get the message can be a little bit cryptic now but not sure this update covers all the case or make it more clear. It may be a bit messy to try to cover all the cases as it would need to infer what the user is trying to pass to the Operator.

I agree, it doesn't. I'm not so sure if this is right message, but maybe just improving the message a little could help. Maybe we can start an issue and close this PR until the discussion takes shape.

I agree an issue with a list of all possible situations ending up there would be useful. After that, we can think of refining the error handling. Basically we need a clear picture of all possible scenarios

I posted the issue #1928 as suggested. I'm going to close this PR so that we can focus on the issue.