Skip to content

CRUD Operations

Learn how to implement CRUD operations for OCPI modules.

Abstract Implementation

The Crud class provides the abstract interface for all CRUD operations in OCPI Python.

Crud

Bases: ABC

Source code in ocpi/core/crud.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class Crud(ABC):
    @abstractmethod
    async def get(cls, module: ModuleID, role: RoleEnum, id, *args, **kwargs) -> Any:
        """Get an object

        :param module: The OCPI module
        :param role: The role of the caller
        :param id: The ID of the object

        :keyword auth_token: (str) The authentication token used by a third
            party
        :keyword version: (VersionNumber) The version number of the caller
            OCPI module
        :keyword party_id: (CiString(3))  The requested party ID
        :keyword country_code: (CiString(2)) The requested Country code
        :keyword token_type: (TokenType) The token type
        :keyword command: (CommandType) The command type of the OCPP command
        :keyword command_data: Request body of command.
        :keyword session_id: (str) Id of the charging profile corresponding
            session.
        :keyword duration: (int) Length of the requested ActiveChargingProfile
            in seconds.
        :keyword response_url: (str) Url where to send result of the action.
        :keyword charging_profile: (SetChargingProfile) ChargingProfile body.
        :keyword session_id: (str) Id of the charging profile corresponding
            session.

        :return: The object data
        :rtype: Any
        """
        pass

    @abstractmethod
    async def list(
        cls, module: ModuleID, role: RoleEnum, filters: dict, *args, **kwargs
    ) -> tuple[list, int, bool]:
        """Get the list of objects

        :param module: The OCPI module
        :param role: The role of the caller
        :param filters: OCPI pagination filters

        :keyword auth_token: (str) The authentication token used by a third
            party
        :keyword version: (VersionNumber) The version number of the caller
            OCPI module
        :keyword party_id: (CiString(3))  The requested party ID
        :keyword country_code: (CiString(2)) The requested Country code

        :return:  Objects list, Total number of objects, if
            it's the last page or not(for pagination)
        :rtype: Tuple[list, int, bool]
        """
        pass

    @abstractmethod
    async def create(
        cls, module: ModuleID, role: RoleEnum, data: dict, *args, **kwargs
    ) -> Any:
        """Create an object

        :param module: The OCPI module
        :param role: The role of the caller
        :param data: The object details

        :keyword auth_token: (str) The authentication token used by
            a third party
        :keyword version: (VersionNumber) The version number of the caller
            OCPI module
        :keyword party_id: (CiString(3))  The requested party ID
        :keyword country_code: (CiString(2)) The requested Country code
        :keyword token_type: (TokenType) The token type
        :keyword command: (CommandType) The command type of the OCPP command
        :keyword query_params: (dict) Charging profile request query params.

        :return: The created object data
        :rtype: Any
        """
        pass

    @abstractmethod
    async def update(
        cls,
        module: ModuleID,
        role: RoleEnum,
        data: dict,
        id: Any,
        *args,
        **kwargs,
    ) -> Any:
        """Update an object

        :param module: The OCPI module
        :param role: The role of the caller
        :param data: The object details
        :param id: The ID of the object

        :keyword auth_token: (str) The authentication token used by a third
            party
        :keyword version: (VersionNumber) The version number of the caller
            OCPI module
        :keyword party_id: (CiString(3))  The requested party ID
        :keyword country_code: (CiString(2)) The requested Country code
        :keyword token_type: (TokenType) The token type
        :keyword session_id: (str) Charging profile corresponding session id.

        :return: The updated object data
        :rtype: Any
        """
        pass

    @abstractmethod
    async def delete(cls, module: ModuleID, role: RoleEnum, id, *args, **kwargs):
        """Delete an object

        :param module: The OCPI module
        :param role: The role of the caller
        :param id: The ID of the object

        :keyword auth_token: (str) The authentication token used by a third
            party
        :keyword version: (VersionNumber) The version number of the caller
            OCPI module
        """
        pass

    @abstractmethod
    async def do(
        cls,
        module: ModuleID,
        role: RoleEnum | None,
        action: Action,
        *args,
        data: dict | None = None,
        **kwargs,
    ) -> Any:
        """Do an action (non-CRUD)

        :param module: The OCPI module
        :param role: The role of the caller
        :param action: The action type
        :param data: The data required for the action

        :keyword auth_token: (str) The authentication token used by a third
            party
        :keyword version: (VersionNumber) The version number of the caller
            OCPI module
        :keyword response_url: (str) Response url for actions which require
            sending response.
        :keyword session: (Session) Session of charging profile action.
        :keyword duration: (int) Length of the requested ActiveChargingProfile
            in seconds.
        :keyword command: (CommandType) The command type of the OCPP command
        :keyword charging_profile (SetChargingProfile): Charging profile sent
            to be updated.

        :return: The action result
        :rtype: Any
        """
        pass

Functions

get abstractmethod async 

get(module, role, id, *args, **kwargs)

Get an object

:param module: The OCPI module :param role: The role of the caller :param id: The ID of the object

:keyword auth_token: (str) The authentication token used by a third party :keyword version: (VersionNumber) The version number of the caller OCPI module :keyword party_id: (CiString(3)) The requested party ID :keyword country_code: (CiString(2)) The requested Country code :keyword token_type: (TokenType) The token type :keyword command: (CommandType) The command type of the OCPP command :keyword command_data: Request body of command. :keyword session_id: (str) Id of the charging profile corresponding session. :keyword duration: (int) Length of the requested ActiveChargingProfile in seconds. :keyword response_url: (str) Url where to send result of the action. :keyword charging_profile: (SetChargingProfile) ChargingProfile body. :keyword session_id: (str) Id of the charging profile corresponding session.

:return: The object data :rtype: Any

Source code in ocpi/core/crud.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@abstractmethod
async def get(cls, module: ModuleID, role: RoleEnum, id, *args, **kwargs) -> Any:
    """Get an object

    :param module: The OCPI module
    :param role: The role of the caller
    :param id: The ID of the object

    :keyword auth_token: (str) The authentication token used by a third
        party
    :keyword version: (VersionNumber) The version number of the caller
        OCPI module
    :keyword party_id: (CiString(3))  The requested party ID
    :keyword country_code: (CiString(2)) The requested Country code
    :keyword token_type: (TokenType) The token type
    :keyword command: (CommandType) The command type of the OCPP command
    :keyword command_data: Request body of command.
    :keyword session_id: (str) Id of the charging profile corresponding
        session.
    :keyword duration: (int) Length of the requested ActiveChargingProfile
        in seconds.
    :keyword response_url: (str) Url where to send result of the action.
    :keyword charging_profile: (SetChargingProfile) ChargingProfile body.
    :keyword session_id: (str) Id of the charging profile corresponding
        session.

    :return: The object data
    :rtype: Any
    """
    pass

list abstractmethod async 

list(module, role, filters, *args, **kwargs)

Get the list of objects

:param module: The OCPI module :param role: The role of the caller :param filters: OCPI pagination filters

:keyword auth_token: (str) The authentication token used by a third party :keyword version: (VersionNumber) The version number of the caller OCPI module :keyword party_id: (CiString(3)) The requested party ID :keyword country_code: (CiString(2)) The requested Country code

:return: Objects list, Total number of objects, if it's the last page or not(for pagination) :rtype: Tuple[list, int, bool]

Source code in ocpi/core/crud.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@abstractmethod
async def list(
    cls, module: ModuleID, role: RoleEnum, filters: dict, *args, **kwargs
) -> tuple[list, int, bool]:
    """Get the list of objects

    :param module: The OCPI module
    :param role: The role of the caller
    :param filters: OCPI pagination filters

    :keyword auth_token: (str) The authentication token used by a third
        party
    :keyword version: (VersionNumber) The version number of the caller
        OCPI module
    :keyword party_id: (CiString(3))  The requested party ID
    :keyword country_code: (CiString(2)) The requested Country code

    :return:  Objects list, Total number of objects, if
        it's the last page or not(for pagination)
    :rtype: Tuple[list, int, bool]
    """
    pass

create abstractmethod async 

create(module, role, data, *args, **kwargs)

Create an object

:param module: The OCPI module :param role: The role of the caller :param data: The object details

:keyword auth_token: (str) The authentication token used by a third party :keyword version: (VersionNumber) The version number of the caller OCPI module :keyword party_id: (CiString(3)) The requested party ID :keyword country_code: (CiString(2)) The requested Country code :keyword token_type: (TokenType) The token type :keyword command: (CommandType) The command type of the OCPP command :keyword query_params: (dict) Charging profile request query params.

:return: The created object data :rtype: Any

Source code in ocpi/core/crud.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@abstractmethod
async def create(
    cls, module: ModuleID, role: RoleEnum, data: dict, *args, **kwargs
) -> Any:
    """Create an object

    :param module: The OCPI module
    :param role: The role of the caller
    :param data: The object details

    :keyword auth_token: (str) The authentication token used by
        a third party
    :keyword version: (VersionNumber) The version number of the caller
        OCPI module
    :keyword party_id: (CiString(3))  The requested party ID
    :keyword country_code: (CiString(2)) The requested Country code
    :keyword token_type: (TokenType) The token type
    :keyword command: (CommandType) The command type of the OCPP command
    :keyword query_params: (dict) Charging profile request query params.

    :return: The created object data
    :rtype: Any
    """
    pass

update abstractmethod async 

update(module, role, data, id, *args, **kwargs)

Update an object

:param module: The OCPI module :param role: The role of the caller :param data: The object details :param id: The ID of the object

:keyword auth_token: (str) The authentication token used by a third party :keyword version: (VersionNumber) The version number of the caller OCPI module :keyword party_id: (CiString(3)) The requested party ID :keyword country_code: (CiString(2)) The requested Country code :keyword token_type: (TokenType) The token type :keyword session_id: (str) Charging profile corresponding session id.

:return: The updated object data :rtype: Any

Source code in ocpi/core/crud.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@abstractmethod
async def update(
    cls,
    module: ModuleID,
    role: RoleEnum,
    data: dict,
    id: Any,
    *args,
    **kwargs,
) -> Any:
    """Update an object

    :param module: The OCPI module
    :param role: The role of the caller
    :param data: The object details
    :param id: The ID of the object

    :keyword auth_token: (str) The authentication token used by a third
        party
    :keyword version: (VersionNumber) The version number of the caller
        OCPI module
    :keyword party_id: (CiString(3))  The requested party ID
    :keyword country_code: (CiString(2)) The requested Country code
    :keyword token_type: (TokenType) The token type
    :keyword session_id: (str) Charging profile corresponding session id.

    :return: The updated object data
    :rtype: Any
    """
    pass

delete abstractmethod async 

delete(module, role, id, *args, **kwargs)

Delete an object

:param module: The OCPI module :param role: The role of the caller :param id: The ID of the object

:keyword auth_token: (str) The authentication token used by a third party :keyword version: (VersionNumber) The version number of the caller OCPI module

Source code in ocpi/core/crud.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@abstractmethod
async def delete(cls, module: ModuleID, role: RoleEnum, id, *args, **kwargs):
    """Delete an object

    :param module: The OCPI module
    :param role: The role of the caller
    :param id: The ID of the object

    :keyword auth_token: (str) The authentication token used by a third
        party
    :keyword version: (VersionNumber) The version number of the caller
        OCPI module
    """
    pass

do abstractmethod async 

do(module, role, action, *args, data=None, **kwargs)

Do an action (non-CRUD)

:param module: The OCPI module :param role: The role of the caller :param action: The action type :param data: The data required for the action

:keyword auth_token: (str) The authentication token used by a third party :keyword version: (VersionNumber) The version number of the caller OCPI module :keyword response_url: (str) Response url for actions which require sending response. :keyword session: (Session) Session of charging profile action. :keyword duration: (int) Length of the requested ActiveChargingProfile in seconds. :keyword command: (CommandType) The command type of the OCPP command :keyword charging_profile (SetChargingProfile): Charging profile sent to be updated.

:return: The action result :rtype: Any

Source code in ocpi/core/crud.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@abstractmethod
async def do(
    cls,
    module: ModuleID,
    role: RoleEnum | None,
    action: Action,
    *args,
    data: dict | None = None,
    **kwargs,
) -> Any:
    """Do an action (non-CRUD)

    :param module: The OCPI module
    :param role: The role of the caller
    :param action: The action type
    :param data: The data required for the action

    :keyword auth_token: (str) The authentication token used by a third
        party
    :keyword version: (VersionNumber) The version number of the caller
        OCPI module
    :keyword response_url: (str) Response url for actions which require
        sending response.
    :keyword session: (Session) Session of charging profile action.
    :keyword duration: (int) Length of the requested ActiveChargingProfile
        in seconds.
    :keyword command: (CommandType) The command type of the OCPP command
    :keyword charging_profile (SetChargingProfile): Charging profile sent
        to be updated.

    :return: The action result
    :rtype: Any
    """
    pass

Example Implementation

See the Quick Start Guide for a complete example of implementing CRUD operations with a database interface.