diff --git a/src/django_opfield/fields.py b/src/django_opfield/fields.py index 7471478..ee6f92a 100644 --- a/src/django_opfield/fields.py +++ b/src/django_opfield/fields.py @@ -22,38 +22,46 @@ class OPField(models.CharField): description = "1Password secret" def __init__( - self, vaults: list[str] | None = None, *args: Any, **kwargs: Any + self, + vaults: list[str] | None = None, + secret_name: str | None = None, + *args: Any, + **kwargs: Any, ) -> None: self.vaults = vaults + self.secret_name = secret_name kwargs.setdefault("max_length", 255) super().__init__(*args, **kwargs) self.validators.append(OPURIValidator(vaults=self.vaults)) - @classmethod - def with_secret(cls, *args: Any, **kwargs: Any) -> tuple[OPField, property]: - op_uri = cls(*args, **kwargs) + @override + def contribute_to_class( + self, cls: type[models.Model], name: str, private_only: bool = False + ): + super().contribute_to_class(cls, name, private_only) - def secret_getter(self: models.Model) -> str | None: + def get_secret(self: models.Model) -> str | None: if not app_settings.OP_SERVICE_ACCOUNT_TOKEN: - msg = "OP_SERVICE_ACCOUNT_TOKEN is not set" - raise ValueError(msg) + raise ValueError("OP_SERVICE_ACCOUNT_TOKEN is not set") if shutil.which("op") is None: msg = "The 'op' CLI command is not available" raise OSError(msg) - field_value = getattr(self, op_uri.name) - result = subprocess.run(["op", "read", field_value], capture_output=True) + op_uri = getattr(self, name) + result = subprocess.run(["op", "read", op_uri], capture_output=True) if result.returncode != 0: - err = result.stderr.decode("utf-8") - msg = f"Could not read secret from 1Password: {err}" - raise ValueError(msg) - secret = result.stdout.decode("utf-8").strip() - return secret + raise ValueError( + f"Could not read secret from 1Password: {result.stderr.decode('utf-8')}" + ) + return result.stdout.decode("utf-8").strip() + + def set_secret(self: models.Model, value: str) -> None: + raise NotImplementedError("OPField does not support setting secret value") - def secret_setter(self: models.Model, value: str) -> None: - raise NotImplementedError("OPField does not support setting secret values") + property_name = ( + f"{name}_secret" if self.secret_name is None else self.secret_name + ) - secret = property(secret_getter, secret_setter) - return op_uri, secret + setattr(cls, property_name, property(get_secret, set_secret)) @override def deconstruct(self): diff --git a/tests/models.py b/tests/models.py index db02524..f46c4ad 100644 --- a/tests/models.py +++ b/tests/models.py @@ -6,4 +6,4 @@ class TestModel(models.Model): - op_uri, op_secret = OPField.with_secret() + op_uri = OPField() diff --git a/tests/test_fields.py b/tests/test_fields.py index 57aa3f3..6b2ab07 100644 --- a/tests/test_fields.py +++ b/tests/test_fields.py @@ -49,13 +49,6 @@ def test_init_validator(): assert OPURIValidator in validators -def test_field_return(): - field, secret = OPField.with_secret() - - assert isinstance(field, OPField) - assert isinstance(secret, property) - - def test_deconstruct_default(): field = OPField() @@ -91,7 +84,7 @@ def test_get_secret(mock_run): model = TestModel(op_uri="op://vault/item/field") - secret = model.op_secret + secret = model.op_uri_secret mock_run.assert_called_once_with( ["op", "read", "op://vault/item/field"], capture_output=True @@ -107,7 +100,7 @@ def test_get_secret_no_token(mock_run): model = TestModel(op_uri="op://vault/item/field") with pytest.raises(ValueError) as exc_info: - _ = model.op_secret + _ = model.op_uri_secret assert "OP_SERVICE_ACCOUNT_TOKEN is not set" in str(exc_info.value) @@ -121,7 +114,7 @@ def test_get_secret_error(mock_run): model = TestModel(op_uri="op://vault/item/field") with pytest.raises(ValueError) as exc_info: - _ = model.op_secret + _ = model.op_uri_secret assert "Could not read secret from 1Password" in str(exc_info.value) @@ -132,7 +125,7 @@ def test_get_secret_command_not_available(mock_which, db): model = TestModel(op_uri="op://vault/item/field") with pytest.raises(OSError) as excinfo: - _ = model.op_secret + _ = model.op_uri_secret assert "The 'op' CLI command is not available" in str(excinfo.value) @@ -143,10 +136,10 @@ def test_set_secret_failure(mock_run): model = TestModel(op_uri="op://vault/item/field") with pytest.raises(NotImplementedError) as exc_info: - model.op_secret = "new secret" + model.op_uri_secret = "new secret" model.save() - assert "OPField does not support setting secret values" in str(exc_info.value) + assert "OPField does not support setting secret value" in str(exc_info.value) @pytest.mark.parametrize(