Create a new datasource. The datasource will be henceforth identified by account/container which must be unique or this function will return a 400. This will encrypt the sasToken if it is provided. Parameters ---------- datasource : DataSource The datasource to crea
(datasource: DataSource, user: Optional[dict])
| 102 | |
| 103 | |
| 104 | async def create(datasource: DataSource, user: Optional[dict]) -> DataSource: |
| 105 | """ |
| 106 | Create a new datasource. The datasource will be henceforth identified by account/container which |
| 107 | must be unique or this function will return a 400. |
| 108 | This will encrypt the sasToken if it is provided. |
| 109 | |
| 110 | Parameters |
| 111 | ---------- |
| 112 | datasource : DataSource |
| 113 | The datasource to create. |
| 114 | |
| 115 | Returns |
| 116 | ------- |
| 117 | DataSource |
| 118 | The datasource. |
| 119 | """ |
| 120 | datasource_collection: AgnosticCollection = collection() |
| 121 | if await datasource_exists(datasource.account, datasource.container): |
| 122 | raise Exception("Datasource Already Exists") |
| 123 | if datasource.sasToken: |
| 124 | datasource.sasToken = encrypt(datasource.sasToken) |
| 125 | else: |
| 126 | datasource.sasToken = "" |
| 127 | if datasource.accountKey: |
| 128 | datasource.accountKey = encrypt(datasource.accountKey) |
| 129 | else: |
| 130 | datasource.accountKey = "" |
| 131 | datasource_dict = datasource.dict(by_alias=True, exclude_unset=True) |
| 132 | |
| 133 | if "owners" not in datasource_dict: |
| 134 | datasource_dict["owners"] = [] |
| 135 | if user and user["preferred_username"]: |
| 136 | datasource_dict["owners"].append(user["preferred_username"]) |
| 137 | if "readers" not in datasource_dict: |
| 138 | datasource_dict["readers"] = [] |
| 139 | if "public" not in datasource_dict: |
| 140 | datasource_dict["public"] = True |
| 141 | await datasource_collection.insert_one(datasource_dict) |
| 142 | return datasource |
no test coverage detected