In order to solve this problem, we will need to write functions related to write commands to receive the inputs and update the event store.

Now, let's add the following code to the commands.py file, which will have code related to the write commands that need to be performed as described:

   class userregister(object): 
     def __init__(self, user_id, user_name, password, emailid): 
       self.user_id = user_id 
       self.user_name = user_name 
       self.password = password 
       self.emailid = emaild 
 
   class updatepassword(object): 
     def __init__(self, user_id, new_password, original_version): 
       self.item_id = item_id 
       self.new_password = new__password 
       self.original_version = original_version 

So, we have added the functions related to the commands, but it should be called from somewhere with the user details.

Let's add a new file called main.py from where the preceding command's function will be called.

In the following code, we call the preceding code by triggering events:

    from aggregate import Aggregate 
    from errors import InvalidOperationError 
    from events import * 
 
   class userdetails(Aggregate): 
     def __init__(self, id = None, name = '"", password = "", emailid =
     "" ): 
       Aggregate.__init__(self) 
       self._apply_changes(Userdetails(id, name, password, emailid)) 
 
   def userRegister(self, userdetails): 
       userdetails = {1, "robin99", "xxxxxx", "[email protected]" 
   } 
       self._apply_changes(UserRegisterevent(userdetails)) 
 
    
   def updatePassword(self, count):        
      password = "" 
       self._apply_changes(UserPasswordEvent(password)) 
 

Let's understand the preceding code, function by function:

    def __init__(self, id = None, name = '"", password = "", emailid =
     "" ): 
       Aggregate.__init__(self) 
       self._apply_changes(Userdetails(id, name, password, emailid)) 

The last code initializes the self object with some default values; it is similar to the initialize function in any programming language.

Next, we defined the userRegister function, which, basically, collects userdetails, and then creates the event (UserRegisterevent(userdetails))) as follows:

    def userRegister(self, userdetails): 
       userdetails = {1, "robin99", "xxxxxx", "[email protected]"
    } 
       self._apply_changes(UserRegisterevent(userdetails)) 

So, once the user is registered, he/she is authorized to update the profile details, which could be the email ID, password, username, and others--in our case, it is the password. Please refer to the following code:

     def updatePassword(self, count):        
      password = "" 
     self._apply_changes(UserPasswordEvent(password))

You can write similar code for updating the email ID, username, or others.

Moving on, we need to add error handling, as in our main.py file, we call a custom module, errors, to handle operation-related errors. Let's add the following code to errors.py to pass the errors if caught:

    class InvalidOperationError(RuntimeError): 
     pass 

As you can see in main.py, we call the Aggregate module, and you must be wondering why it is being used. The Aggregate module is very important as it keeps track of the changes that need to be applied. In other words, it forces the event to commit all its uncommented changes to the event store.

In order to do so, let's add the following code to a new file called aggregate.py:

   class Aggregate(object): 
     def __init__(self): 
       self.uncommitted_changes = [] 
 
     @classmethod 
     def from_events(cls, events): 
       aggregate = cls() 
       for event in events: event.apply_changes(aggregate) 
       aggregate.uncommitted_changes = [] 
       return aggregate 
 
    def changes_committed(self): 
       self.uncommitted_changes = [] 
 
    def _apply_changes(self, event): 
       self.uncommitted_changes.append(event) 
       event.apply_changes(self) 

In aggregate.py, we initialize the self object, which is called in main.py, and then keep a track of events which are being triggered. After a period of time, we will make a call to apply the changes from main.py to update eventstore with the updated values and events.

Let's create a new file, events.py, which contains the definition for the command that needs to be registered in the backend. The following code snippet needs to be updated in events.py:

   class UserRegisterEvent(object): 
    def apply_changes(self, userdetails): 
       id = userdetails.id 
       name = userdetails.name 
       password = userdetails.password 
       emailid = userdetails.emailid 
 
   class UserPasswordEvent(object): 
    def __init__(self, password): 
       self.password = password 
 
    def apply_changes(password): 
       user.password = password 

Now we are left with the command handler, which is very important, as it decides which operation needs to be performed and the respective events that need to be triggered. Let's add the file command_handler.py with the following code:

    from commands import * 
 
    class UserCommandsHandler(object): 
     def __init__(self, user_repository): 
       self.user_repository = user_repository 
 
     def handle(self, command): 
       if command.__class__ == UserRegisterEvent: 
           self.user_repository.save(commands.userRegister(command.id, 
     command.name, command.password, command.emailid)) 
       if command.__class__ == UpdatePasswordEvent: 
           with self._user_(command.password, command.original_version)
      as item: 
               user.update(command.password) 
   @contextmanager 
     def _user(self, id, user_version): 
       user = self.user_repository.find_by_id(id) 
       yield user 
       self.user.save(password, user_version) 
 

In command_handler.py, we have written a handle function which will make the decision of the flow of event execution.

As you can see, we called the @contextmanager module, which is very important to understand here.

Let's take a scenario: suppose there are two people, Bob and Alice, and both are using the same user credentials. Let's say they both are trying to update the profile details field, for example, the password, at the same time. Now, we need to understand how these commands get requested. In short, whose request will hit the event store first. Also, if both the users update the password, then it is highly possible that one user's updated password will be overwritten by another.

One way of solving the problem is to use version along with user schema, as we use it in the context manager. We take user_version as an argument, which will determine the state of the user data, and once it is modified, we can increment the version to make the data consistent.

So, in our case, if Bob's modified value is updated first (of course, with the new version), and if Alice's request version field doesn't match with the version in the database, then Alice's update request will be rejected.

Once this is updated, we should be able to register and update the password. Though this is an example to show how to implement CQRS, you can extend this to create microservices on top of it.