Make data-links behave more like 'symlinks' (#9485)

- Closes #9324

distribution/lib/Standard/AWS/0.0.0-dev/src/S3/S3_Data_Link.enso

@@ -1,5 +1,7 @@
 from Standard.Base import all
-from Standard.Base.Enso_Cloud.Data_Link import parse_format
+import Standard.Base.Errors.Illegal_State.Illegal_State
+import Standard.Base.System.Input_Stream.Input_Stream
+from Standard.Base.Enso_Cloud.Data_Link import Data_Link_With_Input_Stream, parse_format
 from Standard.Base.Enso_Cloud.Public_Utils import get_optional_field, get_required_field
 
 import project.AWS_Credential.AWS_Credential
@@ -9,18 +11,30 @@ from project.Internal.Data_Link_Helpers import decode_aws_credential
 ## PRIVATE
 type S3_Data_Link
     ## PRIVATE
-    Value (uri : Text) format (credentials : AWS_Credential)
+    Value (uri : Text) format_json (credentials : AWS_Credential)
 
     ## PRIVATE
     parse json -> S3_Data_Link =
         uri = get_required_field "uri" json expected_type=Text
         auth = decode_aws_credential (get_required_field "auth" json)
-        format = parse_format (get_optional_field "format" json)
-        S3_Data_Link.Value uri format auth
+        format_json = get_optional_field "format" json
+        S3_Data_Link.Value uri format_json auth
 
     ## PRIVATE
     as_file self -> S3_File = S3_File.new self.uri self.credentials
 
     ## PRIVATE
-    read self (on_problems : Problem_Behavior) =
-        self.as_file.read self.format on_problems
+    default_format self -> Any ! Illegal_State =
+        parse_format self.format_json
+
+    ## PRIVATE
+    read self (format = Auto_Detect) (on_problems : Problem_Behavior) =
+        effective_format = if format != Auto_Detect then format else self.default_format
+        self.as_file.read effective_format on_problems
+
+    ## PRIVATE
+    with_input_stream self (open_options : Vector) (action : Input_Stream -> Any) -> Any =
+        self.as_file.with_input_stream open_options action
+
+## PRIVATE
+Data_Link_With_Input_Stream.from (that:S3_Data_Link) = Data_Link_With_Input_Stream.Value that

distribution/lib/Standard/AWS/0.0.0-dev/src/S3/S3_File.enso

@@ -1,9 +1,11 @@
 from Standard.Base import all
+import Standard.Base.Enso_Cloud.Data_Link
 import Standard.Base.Errors.Common.Syntax_Error
 import Standard.Base.Errors.File_Error.File_Error
 import Standard.Base.Errors.Illegal_Argument.Illegal_Argument
 import Standard.Base.Errors.Unimplemented.Unimplemented
 import Standard.Base.Runtime.Context
+import Standard.Base.System.File.Data_Link_Access.Data_Link_Access
 import Standard.Base.System.File.Generic.File_Like.File_Like
 import Standard.Base.System.File.Generic.Writable_File.Writable_File
 import Standard.Base.System.File_Format_Metadata.File_Format_Metadata
@@ -89,21 +91,21 @@ type S3_File
     with_output_stream : Vector File_Access -> (Output_Stream -> Any ! File_Error) -> Any ! File_Error
     with_output_stream self (open_options : Vector) action = if self.is_directory then Error.throw (S3_Error.Error "S3 directory cannot be opened as a stream." self.uri) else
         Context.Output.if_enabled disabled_message="Writing to an S3_File is forbidden as the Output context is disabled." panic=False <|
-            if open_options.contains File_Access.Append then Error.throw (S3_Error.Error "S3 does not support appending to a file. Instead you may read it, modify and then write the new contents." self.uri) else
-                # The exists check is not atomic, but it is the best we can do with S3
-                check_exists = open_options.contains File_Access.Create_New
-                valid_options = [File_Access.Write, File_Access.Create_New, File_Access.Truncate_Existing, File_Access.Create]
-                invalid_options = open_options.filter (Filter_Condition.Is_In valid_options action=Filter_Action.Remove)
-                if invalid_options.not_empty then Error.throw (S3_Error.Error "Unsupported S3 stream options: "+invalid_options.to_display_text self.uri) else
-                    if check_exists && self.exists then Error.throw (File_Error.Already_Exists self) else
-                        # Given that the amount of data written may be large and AWS library does not seem to support streaming it directly, we use a temporary file to store the data.
-                        tmp_file = File.create_temporary_file "s3-tmp"
-                        Panic.with_finalizer tmp_file.delete <|
-                            result = tmp_file.with_output_stream [File_Access.Write] action
-                            # Only proceed if the write succeeded
-                            result.if_not_error <|
-                                (translate_file_errors self <| S3.upload_file tmp_file self.s3_path.bucket self.s3_path.key self.credentials) . if_not_error <|
-                                    result
+            open_as_data_link = (open_options.contains Data_Link_Access.No_Follow . not) && (Data_Link.is_data_link self)
+            if open_as_data_link then Data_Link.write_data_link_as_stream self open_options action else
+                if open_options.contains File_Access.Append then Error.throw (S3_Error.Error "S3 does not support appending to a file. Instead you may read it, modify and then write the new contents." self.uri) else
+                    File_Access.ensure_only_allowed_options "with_output_stream" [File_Access.Write, File_Access.Create_New, File_Access.Truncate_Existing, File_Access.Create, Data_Link_Access.No_Follow] open_options <|
+                        # The exists check is not atomic, but it is the best we can do with S3
+                        check_exists = open_options.contains File_Access.Create_New
+                        if check_exists && self.exists then Error.throw (File_Error.Already_Exists self) else
+                            # Given that the amount of data written may be large and AWS library does not seem to support streaming it directly, we use a temporary file to store the data.
+                            tmp_file = File.create_temporary_file "s3-tmp"
+                            Panic.with_finalizer tmp_file.delete <|
+                                result = tmp_file.with_output_stream [File_Access.Write] action
+                                # Only proceed if the write succeeded
+                                result.if_not_error <|
+                                    (translate_file_errors self <| S3.upload_file tmp_file self.s3_path.bucket self.s3_path.key self.credentials) . if_not_error <|
+                                        result
 
 
     ## PRIVATE
@@ -121,9 +123,11 @@ type S3_File
        if it returns exceptionally).
     with_input_stream : Vector File_Access -> (Input_Stream -> Any ! File_Error) -> Any ! S3_Error | Illegal_Argument
     with_input_stream self (open_options : Vector) action = if self.is_directory then Error.throw (Illegal_Argument.Error "S3 folders cannot be opened as a stream." self.uri) else
-        if (open_options != [File_Access.Read]) then Error.throw (S3_Error.Error "S3 files can only be opened for reading." self.uri) else
-            response_body = translate_file_errors self <| S3.get_object self.s3_path.bucket self.s3_path.key self.credentials delimiter=S3_Path.delimiter
-            response_body.with_stream action
+        open_as_data_link = (open_options.contains Data_Link_Access.No_Follow . not) && (Data_Link.is_data_link self)
+        if open_as_data_link then Data_Link.read_data_link_as_stream self open_options action else
+            File_Access.ensure_only_allowed_options "with_output_stream" [File_Access.Read, Data_Link_Access.No_Follow] open_options <|
+                response_body = translate_file_errors self <| S3.get_object self.s3_path.bucket self.s3_path.key self.credentials delimiter=S3_Path.delimiter
+                response_body.with_stream action
 
     ## ALIAS load, open
        GROUP Standard.Base.Input
@@ -143,14 +147,14 @@ type S3_File
     @format File_Format.default_widget
     read : File_Format -> Problem_Behavior -> Any ! S3_Error
     read self format=Auto_Detect (on_problems=Problem_Behavior.Report_Warning) =
-        _ = on_problems
-        File_Format.handle_format_missing_arguments format <| case format of
-            Auto_Detect -> if self.is_directory then format.read self on_problems else
-                response = translate_file_errors self <| S3.get_object self.s3_path.bucket self.s3_path.key self.credentials delimiter=S3_Path.delimiter
-                response.decode Auto_Detect
-            _ ->
-                metadata = File_Format_Metadata.Value path=self.path name=self.name
-                self.with_input_stream [File_Access.Read] (stream-> format.read_stream stream metadata)
+        if Data_Link.is_data_link self then Data_Link.read_data_link self format on_problems else
+            File_Format.handle_format_missing_arguments format <| case format of
+                Auto_Detect -> if self.is_directory then format.read self on_problems else
+                    response = translate_file_errors self <| S3.get_object self.s3_path.bucket self.s3_path.key self.credentials delimiter=S3_Path.delimiter
+                    response.decode Auto_Detect
+                _ ->
+                    metadata = File_Format_Metadata.Value path=self.path name=self.name
+                    self.with_input_stream [File_Access.Read] (stream-> format.read_stream stream metadata)
 
     ## ALIAS load bytes, open bytes
        ICON data_input

distribution/lib/Standard/Base/0.0.0-dev/src/Data.enso

@@ -3,16 +3,17 @@ import project.Data.Pair.Pair
 import project.Data.Text.Encoding.Encoding
 import project.Data.Text.Text
 import project.Data.Vector.Vector
+import project.Enso_Cloud.Data_Link
 import project.Error.Error
 import project.Errors.File_Error.File_Error
 import project.Errors.Illegal_Argument.Illegal_Argument
 import project.Errors.Problem_Behavior.Problem_Behavior
+import project.Internal.Data_Read_Helpers
 import project.Meta
 import project.Network.HTTP.Header.Header
 import project.Network.HTTP.HTTP
 import project.Network.HTTP.HTTP_Error.HTTP_Error
 import project.Network.HTTP.HTTP_Method.HTTP_Method
-import project.Network.HTTP.Request.Request
 import project.Network.HTTP.Request_Body.Request_Body
 import project.Network.HTTP.Request_Error
 import project.Network.URI.URI
@@ -21,6 +22,7 @@ import project.Runtime.Context
 import project.System.File.File
 import project.System.File.Generic.Writable_File.Writable_File
 from project.Data.Boolean import Boolean, False, True
+from project.Metadata.Choice import Option
 from project.Metadata.Widget import Text_Input
 from project.System.File_Format import Auto_Detect, File_Format
 
@@ -65,8 +67,9 @@ from project.System.File_Format import Auto_Detect, File_Format
 @format File_Format.default_widget
 read : Text | File -> File_Format -> Problem_Behavior -> Any ! File_Error
 read path format=Auto_Detect (on_problems=Problem_Behavior.Report_Warning) = case path of
-    _ : Text -> if (path.starts_with "http://") || (path.starts_with "https://") then fetch path else
+    _ : Text -> if Data_Read_Helpers.looks_like_uri path then Data_Read_Helpers.fetch_following_data_links path format=format else
         read (File.new path) format on_problems
+    uri : URI -> Data_Read_Helpers.fetch_following_data_links uri format=format
     _ -> File.new path . read format on_problems
 
 ## ALIAS load text, open text
@@ -168,8 +171,9 @@ list_directory directory:File (name_filter:(Text | Nothing)=Nothing) recursive:B
      `HTTP_Method.Head`, `HTTP_Method.Delete`, `HTTP_Method.Options`.
      Defaults to `HTTP_Method.Get`.
    - headers: The headers to send with the request. Defaults to an empty vector.
-   - try_auto_parse_response:  If successful should the body be attempted to be
-     parsed to an Enso native object.
+   - format: The format to use for interpreting the response.
+     Defaults to `Auto_Detect`. If `Raw_Response` is selected or if the format
+     cannot be determined automatically, a raw HTTP `Response` will be returned.
 
    > Example
      Read from an HTTP endpoint.
@@ -184,15 +188,10 @@ list_directory directory:File (name_filter:(Text | Nothing)=Nothing) recursive:B
          file = enso_project.data / "spreadsheet.xls"
          Data.fetch URL . body . write file
 @uri Text_Input
-fetch : (URI | Text) -> HTTP_Method -> Vector (Header | Pair Text Text) -> Boolean -> Any ! Request_Error | HTTP_Error
-fetch (uri:(URI | Text)) (method:HTTP_Method=HTTP_Method.Get) (headers:(Vector (Header | Pair Text Text))=[]) (try_auto_parse_response:Boolean=True) =
-    response = HTTP.fetch uri method headers
-    if try_auto_parse_response.not then response.with_materialized_body else
-        ## We cannot catch decoding errors here and fall-back to the raw response
-           body, because as soon as decoding is started, at least part of the
-           input stream may already be consumed, so we cannot easily reconstruct
-           the whole stream.
-        response.decode if_unsupported=response.with_materialized_body
+@format Data_Read_Helpers.format_widget_with_raw_response
+fetch : (URI | Text) -> HTTP_Method -> Vector (Header | Pair Text Text) -> File_Format -> Any ! Request_Error | HTTP_Error
+fetch (uri:(URI | Text)) (method:HTTP_Method=HTTP_Method.Get) (headers:(Vector (Header | Pair Text Text))=[]) (format = Auto_Detect) =
+    Data_Read_Helpers.fetch_following_data_links uri method headers (Data_Read_Helpers.handle_legacy_format "fetch" "format" format)
 
 ## ALIAS http post, upload
    GROUP Output
@@ -207,8 +206,9 @@ fetch (uri:(URI | Text)) (method:HTTP_Method=HTTP_Method.Get) (headers:(Vector (
    - method: The HTTP method to use. Must be one of `HTTP_Method.Post`,
      `HTTP_Method.Put`, `HTTP_Method.Patch`. Defaults to `HTTP_Method.Post`.
    - headers: The headers to send with the request. Defaults to an empty vector.
-   - try_auto_parse:  If successful should the body be attempted to be parsed to
-     an Enso native object.
+   - response_format: The format to use for interpreting the response.
+     Defaults to `Auto_Detect`. If `Raw_Response` is selected or if the format
+     cannot be determined automatically, a raw HTTP `Response` will be returned.
 
    ! Supported Body Types
 
@@ -314,11 +314,11 @@ fetch (uri:(URI | Text)) (method:HTTP_Method=HTTP_Method.Get) (headers:(Vector (
          form_data = Map.from_vector [["key", "val"], ["a_file", test_file]]
          response = Data.post url_post (Request_Body.Form_Data form_data url_encoded=True)
 @uri Text_Input
-post : (URI | Text) -> Request_Body -> HTTP_Method -> Vector (Header | Pair Text Text) -> Boolean -> Any ! Request_Error | HTTP_Error
-post (uri:(URI | Text)) (body:Request_Body=Request_Body.Empty) (method:HTTP_Method=HTTP_Method.Post) (headers:(Vector (Header | Pair Text Text))=[]) (try_auto_parse_response:Boolean=True) =
+@response_format Data_Read_Helpers.format_widget_with_raw_response
+post : (URI | Text) -> Request_Body -> HTTP_Method -> Vector (Header | Pair Text Text) -> File_Format -> Any ! Request_Error | HTTP_Error
+post (uri:(URI | Text)) (body:Request_Body=Request_Body.Empty) (method:HTTP_Method=HTTP_Method.Post) (headers:(Vector (Header | Pair Text Text))=[]) (response_format = Auto_Detect) =
     response = HTTP.post uri body method headers
-    if try_auto_parse_response.not then response.with_materialized_body else
-        response.decode if_unsupported=response.with_materialized_body
+    Data_Read_Helpers.decode_http_response_following_data_links response (Data_Read_Helpers.handle_legacy_format "post" "response_format" response_format)
 
 ## GROUP Input
    ICON data_download
@@ -337,4 +337,17 @@ download : (URI | Text) -> Writable_File -> HTTP_Method -> Vector (Header | Pair
 download (uri:(URI | Text)) file:Writable_File (method:HTTP_Method=HTTP_Method.Get) (headers:(Vector (Header | Pair Text Text))=[]) =
     Context.Output.if_enabled disabled_message="Downloading to a file is forbidden as the Output context is disabled." panic=False <|
         response = HTTP.fetch uri method headers
-        response.write file
+        case Data_Link.is_data_link response.body.metadata of
+            True ->
+                # If the resource was a data link, we follow it, download the target data and try to write it to a file.
+                data_link = Data_Link.interpret_json_as_data_link response.decode_as_json
+                Data_Link.save_data_link_to_file data_link file
+            False ->
+                response.write file
+
+## If the `format` is set to `Raw_Response`, a raw HTTP `Response` is returned
+   that can be then processed further manually.
+type Raw_Response
+    ## PRIVATE
+    get_dropdown_options : Vector Option
+    get_dropdown_options = [Option "Raw HTTP Response" (Meta.get_qualified_type_name Raw_Response)]

distribution/lib/Standard/Base/0.0.0-dev/src/Data/Array.enso

@@ -149,7 +149,7 @@ type Array
     sort self (order = Sort_Direction.Ascending) on=Nothing by=Nothing on_incomparable=Problem_Behavior.Ignore =
         Array_Like_Helpers.sort self order on by on_incomparable
 
-    ## ALIAS first, last, sample, slice, top, head, tail, foot, limit
+    ## ALIAS first, last, sample, slice, top, head, tail, limit
        GROUP Selections
        ICON select_row
        Creates a new `Vector` with only the specified range of elements from the

distribution/lib/Standard/Base/0.0.0-dev/src/Data/Text/Extensions.enso

@@ -942,7 +942,7 @@ Text.repeat : Integer -> Text
 Text.repeat self count=1 =
     0.up_to count . fold "" acc-> _-> acc + self
 
-## ALIAS first, last, left, mid, right, slice, substring, top, head, tail, foot, limit
+## ALIAS first, last, left, mid, right, slice, substring, top, head, tail, limit
    GROUP Selections
    Creates a new Text by selecting the specified range of the input.
 

distribution/lib/Standard/Base/0.0.0-dev/src/Data/Vector.enso

@@ -870,7 +870,7 @@ type Vector a
     slice : Integer -> Integer -> Vector Any
     slice self start end = Array_Like_Helpers.slice self start end
 
-    ## ALIAS first, last, sample, slice, top, head, tail, foot, limit
+    ## ALIAS first, last, sample, slice, top, head, tail, limit
        GROUP Selections
        ICON select_row
        Creates a new `Vector` with only the specified range of elements from the

distribution/lib/Standard/Base/0.0.0-dev/src/Enso_Cloud/Data_Link.enso

@@ -2,20 +2,31 @@ import project.Any.Any
 import project.Data.Json.JS_Object
 import project.Data.Text.Encoding.Encoding
 import project.Data.Text.Text
+import project.Data.Vector.Vector
 import project.Enso_Cloud.Enso_Secret.Enso_Secret
+import project.Enso_Cloud.Errors.Missing_Data_Link_Library
 import project.Error.Error
+import project.Errors.Common.No_Such_Conversion
+import project.Errors.Illegal_Argument.Illegal_Argument
 import project.Errors.Illegal_State.Illegal_State
 import project.Errors.Problem_Behavior.Problem_Behavior
 import project.Errors.Unimplemented.Unimplemented
 import project.Meta
 import project.Nothing.Nothing
+import project.Panic.Panic
 import project.System.File.File
-import project.System.File.Generic.Writable_File.Writable_File
+import project.System.File.File_Access.File_Access
+import project.System.File.Data_Link_Access.Data_Link_Access
 import project.System.File_Format.Auto_Detect
 import project.System.File_Format.Infer
 import project.System.File_Format.JSON_Format
 import project.System.File_Format_Metadata.File_Format_Metadata
+import project.System.File_Format_Metadata.Content_Type_Metadata
+import project.System.File.Generic.File_Like.File_Like
+import project.System.File.Generic.Writable_File.Writable_File
 import project.System.Input_Stream.Input_Stream
+from project.Data.Boolean import Boolean, False, True
+from project.Data.Text.Extensions import all
 from project.Enso_Cloud.Public_Utils import get_required_field
 
 polyglot java import org.enso.base.enso_cloud.DataLinkSPI
@@ -25,53 +36,173 @@ polyglot java import org.enso.base.file_format.FileFormatSPI
    A file format for reading data links.
 type Data_Link_Format
     ## PRIVATE
-       If the File_Format supports reading from the file, return a configured instance.
-    for_read : File_Format_Metadata -> Data_Link_Format | Nothing
-    for_read file:File_Format_Metadata =
-        case file.guess_extension of
-            ".datalink" -> Data_Link_Format
-            _ -> Nothing
+       Reads the raw configuration data of a data-link.
+    read_config (file : File_Like) -> JS_Object =
+        text = Data_Link_Format.read_raw_config file
+        text.parse_json
 
     ## PRIVATE
-       Currently writing data links is not supported.
-    for_file_write : Writable_File -> Nothing
-    for_file_write file =
-        _ = file
-        Nothing
+       Writes a data-link configuration to a file.
+
+       Arguments:
+       - file: The file to write the configuration to.
+       - config: The configuration to write to the file.
+       - replace_existing: A flag specifying if the operation should replace an
+         existing file. By default, the operation will fail if the file already
+         exists.
+       - skip_validation: A flag that allows to skip validation. By default,
+         before writing the config we try to parse it to ensure that it
+         represents a valid data-link. In some cases (e.g. testing), we may want
+         to skip that.
+    write_config (file : Writable_File) (config : JS_Object) (replace_existing : Boolean = False) (skip_validation : Boolean = False) =
+        checked = if skip_validation.not then Data_Link_Format.validate_config config
+        checked.if_not_error <|
+            Data_Link_Format.write_raw_config file config.to_json replace_existing
 
     ## PRIVATE
-       We currently don't display data-link as a specific format.
-       It relies on `Auto_Detect`.
-    get_dropdown_options = []
+         Reads the raw configuration data of a data-link, as plain text.
+
+         This is should mostly be used for testing, `read_config` is preferred for normal use.
+
+         Arguments:
+         - file: The file to read the configuration from.
+    read_raw_config (file : File_Like) -> Text =
+        if is_data_link file . not then
+            Panic.throw (Illegal_Argument.Error "Data_Link_Format should only be used for reading config of Data Links, but "+file.to_display_text+" is not a Data Link.")
+        options = [File_Access.Read, Data_Link_Access.No_Follow]
+        bytes = file.underlying.with_input_stream options input_stream->
+            input_stream.read_all_bytes
+        Text.from_bytes bytes data_link_encoding on_problems=Problem_Behavior.Report_Error
 
     ## PRIVATE
-       Implements the `File.read` for this `File_Format`
-    read : File -> Problem_Behavior -> Any
-    read self file on_problems =
-        json = JSON_Format.read file on_problems
-        read_datalink json on_problems
+       Writes raw data as the data-link configuration.
+
+       This is should mostly be used for testing, `write_config` is preferred for normal use.
+
+       Arguments:
+       - file: The file to write the configuration to.
+       - raw_content: The raw data to write to the file.
+       - replace_existing: A flag specifying if the operation should replace an
+         existing file. By default, the operation will fail if the file already
+         exists.
+    write_raw_config (file : Writable_File) (raw_content : Text) (replace_existing : Boolean = False) =
+        if is_data_link file . not then
+            Panic.throw (Illegal_Argument.Error "Data_Link_Format should only be used for writing config to Data Links, but "+file.file.to_display_text+" is not a Data Link.")
+        exist_options = if replace_existing then [File_Access.Create, File_Access.Truncate_Existing] else [File_Access.Create_New]
+        options = exist_options + [File_Access.Write, Data_Link_Access.No_Follow]
+
+        bytes = raw_content.bytes data_link_encoding on_problems=Problem_Behavior.Report_Error
+        r = bytes.if_not_error <| file.with_output_stream options output_stream->
+            output_stream.write_bytes bytes
+        r.if_not_error file.file
 
     ## PRIVATE
-       Implements decoding the format from a stream.
-    read_stream : Input_Stream -> File_Format_Metadata -> Any
-    read_stream self stream:Input_Stream (metadata : File_Format_Metadata) =
-        json = JSON_Format.read_stream stream metadata
-        read_datalink json Problem_Behavior.Report_Error
+       Checks if the config represents a valid data-link.
+
+       If the library providing the data-link is not imported, this function
+       will fail with `Missing_Data_Link_Library`, even if the config would be
+       valid.
+    validate_config (config : JS_Object) -> Nothing ! Missing_Data_Link_Library | Illegal_State =
+        interpret_json_as_data_link config . if_not_error Nothing
 
 ## PRIVATE
-interpret_json_as_datalink json =
+   An interface for a data link description.
+type Data_Link
+    ## PRIVATE
+       Reads a data link and interprets it using the provided format.
+       If the format is `Auto_Detect` (default), a default format provided by the data link is used, if available.
+    read self (format = Auto_Detect) (on_problems : Problem_Behavior = Problem_Behavior.Report_Error) -> Any =
+        _ = [format, on_problems]
+        Unimplemented.throw "This is an interface only."
+
+## PRIVATE
+   A type class representing a data link that can be opened as a stream.
+
+   It requires the underlying data link to provide a `with_input_stream` method.
+type Data_Link_With_Input_Stream
+    ## PRIVATE
+    Value underlying
+
+    ## PRIVATE
+       Opens the data pointed at by the data link as a raw stream.
+    with_input_stream self (open_options : Vector) (action : Input_Stream -> Any) -> Any =
+        self.underlying.with_input_stream open_options action
+
+    ## PRIVATE
+       Creates a `Data_Link_With_Input_Stream` from a data link instance, if
+       that data link supports streaming. If it does not, an error is thrown.
+    find data_link_instance (~if_not_supported = (Error.throw (Illegal_Argument.Error "The "+(data_link_name data_link_instance)+" cannot be opened as a stream."))) -> Data_Link_With_Input_Stream ! Illegal_Argument =
+        handle_no_conversion _ =
+            if_not_supported
+        Panic.catch No_Such_Conversion (Data_Link_With_Input_Stream.from data_link_instance) handle_no_conversion
+
+## PRIVATE
+   All data-link config files should be saved with UTF-8 encoding.
+data_link_encoding = Encoding.utf_8
+
+## PRIVATE
+data_link_content_type = "application/x-enso-datalink"
+
+## PRIVATE
+data_link_extension = ".datalink"
+
+## PRIVATE
+   Checks if the given file is a data-link.
+is_data_link (file_metadata : File_Format_Metadata) -> Boolean =
+    content_type_matches = case file_metadata.interpret_content_type of
+        content_type : Content_Type_Metadata ->
+            content_type.base_type == data_link_content_type
+        _ -> False
+
+    # If the content type matches, it is surely a data link.
+    if content_type_matches then True else
+        ## If the content type does not match, we check the extension even if _different content type was provided_.
+           That is because many HTTP servers will not understand data links and may return a data link with
+           a content type like `text/plain` or `application/json`. We still want to treat the file as a data link
+           if its extension is correct.
+        case file_metadata.guess_extension of
+            extension : Text ->
+                extension == data_link_extension
+            Nothing -> False
+
+## PRIVATE
+interpret_json_as_data_link json =
     typ = get_required_field "type" json expected_type=Text
     case DataLinkSPI.findDataLinkType typ of
         Nothing ->
             library_name = get_required_field "libraryName" json expected_type=Text
-            Error.throw (Illegal_State.Error "The data link for "+typ+" is provided by the library "+library_name+" which is not loaded. Please import the library, and if necessary, restart the project.")
+            Error.throw (Missing_Data_Link_Library.Error library_name typ)
         data_link_type ->
             data_link_type.parse json
 
 ## PRIVATE
-read_datalink json on_problems =
-    data_link_instance = interpret_json_as_datalink json
-    data_link_instance.read on_problems
+read_data_link (file : File_Like) format (on_problems : Problem_Behavior) =
+    json = Data_Link_Format.read_config file
+    data_link_instance = interpret_json_as_data_link json
+    data_link_instance.read format on_problems
+
+## PRIVATE
+read_data_link_as_stream (file : File_Like) (open_options : Vector) (f : Input_Stream -> Any) =
+    json = Data_Link_Format.read_config file
+    data_link_instance = interpret_json_as_data_link json
+    data_link_with_input_stream = Data_Link_With_Input_Stream.find data_link_instance
+    data_link_with_input_stream.with_input_stream open_options f
+
+## PRIVATE
+write_data_link (file : File_Like) format (on_problems : Problem_Behavior) =
+    _ = [file, format, on_problems]
+    Unimplemented.throw "Writing data links is not yet supported."
+
+## PRIVATE
+write_data_link_as_stream (file : File_Like) (open_options : Vector) (f : Input_Stream -> Any) =
+    _ = [file, open_options, f]
+    Unimplemented.throw "Writing data links is not yet supported."
+
+## PRIVATE
+save_data_link_to_file data_link_instance (target_file : Writable_File) =
+    data_link_with_input_stream = Data_Link_With_Input_Stream.find data_link_instance if_not_supported=(Error.throw (Illegal_Argument.Error "The "+(data_link_name data_link_instance)+" cannot be saved to a file."))
+    data_link_with_input_stream.with_input_stream [File_Access.Read] input_stream->
+        input_stream.write_to_file target_file
 
 ## PRIVATE
 parse_secure_value json -> Text | Enso_Secret =
@@ -86,7 +217,7 @@ parse_secure_value json -> Text | Enso_Secret =
         _ -> Error.throw (Illegal_State.Error "Parsing a secure value failed. Expected either a string or an object representing a secret, but got "+(Meta.type_of json . to_display_text)+".")
 
 ## PRIVATE
-parse_format json = case json of
+parse_format (json : Any) -> Any ! Illegal_State = case json of
     Nothing -> Auto_Detect
     _ : JS_Object -> case get_required_field "subType" json of
         "default" -> Auto_Detect
@@ -97,3 +228,8 @@ parse_format json = case json of
         other ->
             Error.throw (Illegal_State.Error "Expected `subType` to be a string, but got: "+other.to_display_text+".")
     other -> Error.throw (Illegal_State.Error "Unexpected value inside of a data-link `format` field: "+other.to_display_text+".")
+
+## PRIVATE
+   Returns a human readable name of the data link type, based on its type.
+data_link_name data_link_instance =
+    Meta.type_of data_link_instance . to_display_text . replace "_" " "

distribution/lib/Standard/Base/0.0.0-dev/src/Enso_Cloud/Enso_File.enso

@@ -9,6 +9,7 @@ import project.Data.Text.Text_Sub_Range.Text_Sub_Range
 import project.Data.Time.Date_Time.Date_Time
 import project.Data.Time.Date_Time_Formatter.Date_Time_Formatter
 import project.Data.Vector.Vector
+import project.Enso_Cloud.Data_Link
 import project.Enso_Cloud.Errors.Enso_Cloud_Error
 import project.Enso_Cloud.Internal.Enso_Path.Enso_Path
 import project.Enso_Cloud.Internal.Utils
@@ -23,7 +24,9 @@ import project.Errors.Unimplemented.Unimplemented
 import project.Network.HTTP.HTTP
 import project.Network.HTTP.HTTP_Method.HTTP_Method
 import project.Nothing.Nothing
+import project.Runtime
 import project.Runtime.Context
+import project.System.File.Data_Link_Access.Data_Link_Access
 import project.System.File.File_Access.File_Access
 import project.System.File.Generic.Writable_File.Writable_File
 import project.System.File_Format_Metadata.File_Format_Metadata
@@ -31,7 +34,6 @@ import project.System.Input_Stream.Input_Stream
 import project.System.Output_Stream.Output_Stream
 from project.Data.Boolean import Boolean, False, True
 from project.Data.Text.Extensions import all
-from project.Enso_Cloud.Data_Link import read_datalink
 from project.Enso_Cloud.Public_Utils import get_required_field
 from project.System.File_Format import Auto_Detect, Bytes, File_Format, Plain_Text_Format
 from project.System.File.Generic.File_Write_Strategy import generic_copy
@@ -185,10 +187,20 @@ type Enso_File
        The created stream is automatically closed when `action` returns (even
        if it returns exceptionally).
     with_input_stream : Vector File_Access -> (Input_Stream -> Any ! File_Error) -> Any ! File_Error | Illegal_Argument
-    with_input_stream self (open_options : Vector) action = if self.asset_type != Enso_Asset_Type.File then Error.throw (Illegal_Argument.Error "Only files can be opened as a stream.") else
-        if (open_options !=  [File_Access.Read]) then Error.throw (Illegal_Argument.Error "Files can only be opened for reading.") else
-            response = HTTP.fetch (get_download_url_for_file self) HTTP_Method.Get []
-            response.if_not_error <| response.body.with_stream action
+    with_input_stream self (open_options : Vector) action =
+        open_as_data_link = (open_options.contains Data_Link_Access.No_Follow . not) && (Data_Link.is_data_link self)
+        if open_as_data_link then Data_Link.read_data_link_as_stream self open_options action else
+            File_Access.ensure_only_allowed_options "with_input_stream" [File_Access.Read, Data_Link_Access.No_Follow] open_options <|
+                uri = case self.asset_type of
+                    Enso_Asset_Type.File ->
+                        get_download_url_for_file self
+                    Enso_Asset_Type.Data_Link ->
+                        Runtime.assert (open_options.contains Data_Link_Access.No_Follow)
+                        self.internal_uri
+                    _ ->
+                        Error.throw (Illegal_Argument.Error "Only files can be opened as a stream.")
+                response = HTTP.fetch uri HTTP_Method.Get []
+                response.if_not_error <| response.body.with_stream action
 
     ## ALIAS load, open
        GROUP Input
@@ -212,7 +224,8 @@ type Enso_File
         Enso_Asset_Type.Secret -> Error.throw (Illegal_Argument.Error "Secrets cannot be read directly.")
         Enso_Asset_Type.Data_Link ->
             json = Utils.http_request_as_json HTTP_Method.Get self.internal_uri
-            read_datalink json on_problems
+            datalink = Data_Link.interpret_json_as_data_link json
+            datalink.read format on_problems
         Enso_Asset_Type.Directory -> if format == Auto_Detect then self.list else Error.throw (Illegal_Argument.Error "Directories can only be read using the Auto_Detect format.")
         Enso_Asset_Type.File -> File_Format.handle_format_missing_arguments format <| case format of
             Auto_Detect ->
@@ -220,9 +233,7 @@ type Enso_File
                 if real_format == Nothing then Error.throw (File_Error.Unsupported_Type self) else
                     self.read real_format on_problems
             _ ->
-                # TODO this is just a placeholder, until we implement the proper path
-                path = "enso://"+self.id
-                metadata = File_Format_Metadata.Value path=path name=self.name
+                metadata = File_Format_Metadata.from self
                 self.with_input_stream [File_Access.Read] (stream-> format.read_stream stream metadata)
 
     ## ALIAS load bytes, open bytes
@@ -374,7 +385,17 @@ Enso_Asset_Type.from (that:Text) = case that of
     _ -> Error.throw (Illegal_Argument.Error "Invalid asset type: "+that.pretty+".")
 
 ## PRIVATE
-File_Format_Metadata.from (that:Enso_File) = File_Format_Metadata.Value Nothing that.name (that.extension.catch _->Nothing)
+File_Format_Metadata.from (that:Enso_File) =
+    # TODO this is just a placeholder, until we implement the proper path
+    path = Nothing
+    case that.asset_type of
+        Enso_Asset_Type.Data_Link ->
+            File_Format_Metadata.Value path=path name=that.name content_type=Data_Link.data_link_content_type
+        Enso_Asset_Type.Directory ->
+            File_Format_Metadata.Value path=path name=that.name extension=(that.extension.catch _->Nothing)
+        Enso_Asset_Type.File ->
+            File_Format_Metadata.Value path=path name=that.name extension=(that.extension.catch _->Nothing)
+        _ -> Error.throw (Illegal_Argument.Error "`File_Format_Metadata` is not available for: "+self.asset_type.to_text+".")
 
 ## PRIVATE
    Fetches the basic information about a file from the Cloud endpoint.

distribution/lib/Standard/Base/0.0.0-dev/src/Enso_Cloud/Errors.enso

@@ -47,3 +47,13 @@ type Enso_Cloud_Error
         Enso_Cloud_Error.Invalid_Response_Payload cause -> "Internal error: A response from Enso Cloud could not be parsed: " + cause.to_display_text
         Enso_Cloud_Error.Unauthorized -> "Enso Cloud credentials file was found, but the service responded with 401 Unauthorized. You may try logging in again and restarting the workflow."
         Enso_Cloud_Error.Connection_Error cause -> "Error connecting to Enso Cloud: " + cause.to_display_text
+
+## PRIVATE
+type Missing_Data_Link_Library
+    ## PRIVATE
+    Error (library_name : Text) (data_link_type : Text)
+
+    ## PRIVATE
+    to_display_text : Text
+    to_display_text self =
+        "The data link for "+self.data_link_type+" is provided by the library "+self.library_name+" which is not loaded. Please import the library, and if necessary, restart the project."

distribution/lib/Standard/Base/0.0.0-dev/src/Errors/File_Error.enso

@@ -48,7 +48,7 @@ type File_Error
 
     ## Indicates that the given file is corrupted, i.e. the data it contains
        is not in the expected format.
-    Corrupted_Format (file : File_Like) (message : Text) (cause : Any | Nothing = Nothing)
+    Corrupted_Format (file : File_Like | Nothing) (message : Text) (cause : Any | Nothing = Nothing)
 
     ## PRIVATE
        Convert the File error to a human-readable format.
@@ -66,7 +66,9 @@ type File_Error
                 (Meta.meta format).constructor.name
             format_name = Panic.catch No_Such_Conversion (File_Like.from format . name) (_->name_from_constructor)
             "Values of type "+data_type.to_text+" cannot be written as format "+format_name.to_text+"."
-        File_Error.Corrupted_Format file msg _ -> "The file at " + file.path + " is corrupted: " + msg
+        File_Error.Corrupted_Format file msg _ ->
+            at_part = if file.is_nothing then "" else " at " + file.path
+            "The file"+at_part+" is corrupted: " + msg
 
     ## PRIVATE
 

distribution/lib/Standard/Base/0.0.0-dev/src/Internal/Data_Read_Helpers.enso

@@ -0,0 +1,62 @@
+private
+
+import project.Data.Text.Case_Sensitivity.Case_Sensitivity
+import project.Data.Text.Text
+import project.Data.Vector.Vector
+import project.Enso_Cloud.Data_Link
+import project.Errors.Deprecated.Deprecated
+import project.Errors.Problem_Behavior.Problem_Behavior
+import project.Metadata.Display
+import project.Metadata.Widget
+import project.Network.HTTP.HTTP
+import project.Network.HTTP.HTTP_Method.HTTP_Method
+import project.Network.URI.URI
+import project.Warning.Warning
+from project.Data.Boolean import Boolean, False, True
+from project.Data.Text.Extensions import all
+from project.Data import Raw_Response
+from project.Metadata.Choice import Option
+from project.Metadata.Widget import Single_Choice
+from project.System.File_Format import Auto_Detect, format_types
+
+
+## PRIVATE
+looks_like_uri path:Text -> Boolean =
+    (path.starts_with "http://" Case_Sensitivity.Insensitive) || (path.starts_with "https://" Case_Sensitivity.Insensitive)
+
+## PRIVATE
+   A common implementation for fetching a resource and decoding it,
+   following encountered data links.
+fetch_following_data_links (uri:URI) (method:HTTP_Method = HTTP_Method.Get) (headers:Vector = []) format =
+    response = HTTP.fetch uri method headers
+    decode_http_response_following_data_links response format
+
+## PRIVATE
+   Decodes a HTTP response, handling data link access.
+decode_http_response_following_data_links response format =
+    # If Raw_Response is requested, we ignore data link handling.
+    if format == Raw_Response then response.with_materialized_body else
+        case Data_Link.is_data_link response.body.metadata of
+            True ->
+                data_link = Data_Link.interpret_json_as_data_link response.decode_as_json
+                data_link.read format Problem_Behavior.Report_Error
+            False ->
+                response.decode format=format if_unsupported=response.with_materialized_body
+
+## PRIVATE
+format_widget_with_raw_response -> Widget =
+    options = ([Auto_Detect, Raw_Response] + format_types).flat_map .get_dropdown_options
+    Single_Choice display=Display.When_Modified values=options
+
+## PRIVATE
+   A helper method that handles the old-style invocation of `Data.fetch` and `Data.post`.
+   Before the introduction of the `format` parameter, these methods had a
+   `try_auto_parse_result` parameter taking a Boolean at the same position.
+   To ensure old code relying on positional arguments still works, we have special handling for the Boolean case.
+   This 'migration' will not work unfortunately if the argument was named.
+handle_legacy_format (method_name : Text) (new_argument_name : Text) format = case format of
+    try_auto_parse_result : Boolean ->
+        new_format = if try_auto_parse_result then Auto_Detect else Raw_Response
+        warning = Deprecated.Warning "Standard.Base.Data" method_name "Deprecated: The `try_auto_parse_result` argument was replaced with `"+new_argument_name+"`. `True` becomes `Auto_Detect` and `False` becomes `Raw_Response`."
+        Warning.attach warning new_format
+    _ -> format

distribution/lib/Standard/Base/0.0.0-dev/src/Main.enso

@@ -97,6 +97,7 @@ from project.Data.Json.Extensions import all
 from project.Data.Range.Extensions import all
 from project.Data.Text.Extensions import all
 from project.Data.Text.Regex import regex
+from project.Data import Raw_Response
 from project.Errors.Problem_Behavior.Problem_Behavior import all
 from project.Meta.Enso_Project import enso_project
 from project.Network.Extensions import all
@@ -199,6 +200,7 @@ from project.Data.Range.Extensions export all
 from project.Data.Statistics export all hiding to_moment_statistic, wrap_java_call, calculate_correlation_statistics, calculate_spearman_rank, calculate_correlation_statistics_matrix, compute_fold, empty_value, is_valid
 from project.Data.Text.Extensions export all
 from project.Data.Text.Regex export regex
+from project.Data export Raw_Response
 from project.Errors.Problem_Behavior.Problem_Behavior export all
 from project.Function export all
 from project.Meta.Enso_Project export enso_project

distribution/lib/Standard/Base/0.0.0-dev/src/Network/Extensions.enso

@@ -4,11 +4,12 @@ import project.Data.Pair.Pair
 import project.Data.Text.Text
 import project.Data.Vector.Vector
 import project.Errors.Common.Syntax_Error
+import project.Internal.Data_Read_Helpers
 import project.Network.HTTP.Header.Header
 import project.Network.HTTP.HTTP_Method.HTTP_Method
 import project.Network.HTTP.Request_Body.Request_Body
 import project.Network.URI.URI
-from project.Data.Boolean import Boolean, False, True
+from project.System.File_Format import Auto_Detect, File_Format
 
 ## ALIAS parse_uri, uri from text
    GROUP Conversions
@@ -38,11 +39,13 @@ Text.to_uri self = URI.parse self
      `HTTP_Method.Head`, `HTTP_Method.Delete`, `HTTP_Method.Options`.
      Defaults to `HTTP_Method.Get`.
    - headers: The headers to send with the request. Defaults to an empty vector.
-   - try_auto_parse_response:  If successful should the body be attempted to be
-     parsed to an Enso native object.
-URI.fetch : HTTP_Method -> Vector (Header | Pair Text Text) -> Boolean -> Any
-URI.fetch self (method:HTTP_Method=HTTP_Method.Get) headers=[] try_auto_parse_response=True =
-    Data.fetch self method headers try_auto_parse_response
+   - format: The format to use for interpreting the response.
+     Defaults to `Auto_Detect`. If `Raw_Response` is selected or if the format
+     cannot be determined automatically, a raw HTTP `Response` will be returned.
+@format Data_Read_Helpers.format_widget_with_raw_response
+URI.fetch : HTTP_Method -> Vector (Header | Pair Text Text) -> File_Format -> Any
+URI.fetch self (method:HTTP_Method=HTTP_Method.Get) headers=[] format=Auto_Detect =
+    Data.fetch self method headers format
 
 ## ALIAS upload, http post
    GROUP Output
@@ -56,8 +59,9 @@ URI.fetch self (method:HTTP_Method=HTTP_Method.Get) headers=[] try_auto_parse_re
    - method: The HTTP method to use. Must be one of `HTTP_Method.Post`,
      `HTTP_Method.Put`, `HTTP_Method.Patch`. Defaults to `HTTP_Method.Post`.
    - headers: The headers to send with the request. Defaults to an empty vector.
-   - try_auto_parse_response:  If successful should the body be attempted to be
-     parsed to an Enso native object.
+   - response_format: The format to use for interpreting the response.
+     Defaults to `Auto_Detect`. If `Raw_Response` is selected or if the format
+     cannot be determined automatically, a raw HTTP `Response` will be returned.
 
    ! Specifying Content Types
 
@@ -81,6 +85,7 @@ URI.fetch self (method:HTTP_Method=HTTP_Method.Get) headers=[] try_auto_parse_re
      - Text: shorthand for `Request_Body.Text that_text`.
      - File: shorthand for `Request_Body.Binary that_file`.
      - Any other Enso object: shorthand for `Request_Body.Json that_object`.
-URI.post : Request_Body -> HTTP_Method -> Vector (Header | Pair Text Text) -> Boolean -> Any
-URI.post self (body:Request_Body=Request_Body.Empty) (method:HTTP_Method=HTTP_Method.Post) (headers:(Vector (Header | Pair Text Text))=[]) (try_auto_parse_response:Boolean=True) =
-    Data.post self body method headers try_auto_parse_response
+@response_format Data_Read_Helpers.format_widget_with_raw_response
+URI.post : Request_Body -> HTTP_Method -> Vector (Header | Pair Text Text) -> File_Format -> Any
+URI.post self (body:Request_Body=Request_Body.Empty) (method:HTTP_Method=HTTP_Method.Post) (headers:(Vector (Header | Pair Text Text))=[]) (response_format = Auto_Detect) =
+    Data.post self body method headers response_format

distribution/lib/Standard/Base/0.0.0-dev/src/Network/HTTP/HTTP_Fetch_Data_Link.enso

@@ -1,29 +1,46 @@
+import project.Any.Any
 import project.Data.Text.Text
+import project.Data.Vector.Vector
 import project.Errors.Problem_Behavior.Problem_Behavior
 import project.Network.HTTP.HTTP
 import project.Network.HTTP.HTTP_Method.HTTP_Method
 import project.Network.HTTP.Request.Request
 import project.Nothing.Nothing
-from project.Enso_Cloud.Data_Link import parse_format, parse_secure_value
+import project.System.File.Data_Link_Access.Data_Link_Access
+import project.System.File.File_Access.File_Access
+import project.System.File_Format.Auto_Detect
+import project.System.Input_Stream.Input_Stream
+from project.Data.Boolean import Boolean, False, True
+from project.Enso_Cloud.Data_Link import Data_Link_With_Input_Stream, parse_format, parse_secure_value
 from project.Enso_Cloud.Public_Utils import get_optional_field, get_required_field
 
 ## PRIVATE
 type HTTP_Fetch_Data_Link
     ## PRIVATE
-    Value (request : Request) format
+    Value (request : Request) format_json
 
     ## PRIVATE
     parse json -> HTTP_Fetch_Data_Link =
         uri = get_required_field "uri" json expected_type=Text
         method = HTTP_Method.from (get_required_field "method" json expected_type=Text)
-        format = parse_format (get_optional_field "format" json)
+        format_json = get_optional_field "format" json
         # TODO headers
         headers = []
         request = Request.new method uri headers
-        HTTP_Fetch_Data_Link.Value request format
+        HTTP_Fetch_Data_Link.Value request format_json
 
     ## PRIVATE
-    read self (on_problems : Problem_Behavior) =
+    read self (format = Auto_Detect) (on_problems : Problem_Behavior) =
         _ = on_problems
+        effective_format = if format != Auto_Detect then format else parse_format self.format_json
         response = HTTP.new.request self.request
-        response.decode self.format
+        response.decode effective_format
+
+    ## PRIVATE
+    with_input_stream self (open_options : Vector) (action : Input_Stream -> Any) -> Any =
+        File_Access.ensure_only_allowed_options "with_output_stream" [File_Access.Read] open_options <|
+            response = HTTP.new.request self.request
+            response.body.with_stream action
+
+## PRIVATE
+Data_Link_With_Input_Stream.from (that:HTTP_Fetch_Data_Link) = Data_Link_With_Input_Stream.Value that

distribution/lib/Standard/Base/0.0.0-dev/src/Network/HTTP/Response.enso

@@ -125,7 +125,8 @@ type Response
        ICON data_input
        Uses the format to decode the body.
        If using `Auto_Detect`, the content-type will be used to determine the
-       format.
+       format. If the format cannot be detected automatically, `if_unsupported`
+       is returned.
     @format decode_format_selector
     decode : File_Format -> Any -> Any
     decode self format=Auto_Detect ~if_unsupported=Throw_Unsupported_Error =

distribution/lib/Standard/Base/0.0.0-dev/src/Network/HTTP/Response_Body.enso

@@ -115,7 +115,8 @@ type Response_Body
 
        Arguments:
        - format: The format to use to decode the body.
-       - if_unsupported: Specifies how to proceed if the format is not supported.
+       - if_unsupported: Specifies how to proceed if `Auto_Detect` was selected
+         but the format could not be determined.
     @format decode_format_selector
     decode : File_Format -> Any -> Any
     decode self format=Auto_Detect ~if_unsupported=Throw_Unsupported_Error =
@@ -203,7 +204,7 @@ type Response_Body
         self.with_stream body_stream->
             file.write on_existing_file output_stream->
                 r = output_stream.write_stream body_stream
-                r.if_not_error file
+                r.if_not_error file.file
 
 ## PRIVATE
 can_decode : File_Format -> Boolean

distribution/lib/Standard/Base/0.0.0-dev/src/System/File.enso

@@ -10,6 +10,7 @@ import project.Data.Text.Text
 import project.Data.Text.Text_Sub_Range.Text_Sub_Range
 import project.Data.Time.Date_Time.Date_Time
 import project.Data.Vector.Vector
+import project.Enso_Cloud.Data_Link
 import project.Error.Error
 import project.Errors.Common.Dry_Run_Operation
 import project.Errors.Common.Type_Error
@@ -23,6 +24,7 @@ import project.Nothing.Nothing
 import project.Panic.Panic
 import project.Runtime.Context
 import project.Runtime.Managed_Resource.Managed_Resource
+import project.System.File.Data_Link_Access.Data_Link_Access
 import project.System.File.File_Access.File_Access
 import project.System.File.File_Permissions.File_Permissions
 import project.System.File.Generic.File_Like.File_Like
@@ -184,15 +186,17 @@ type File
 
             ## Re-wrap the File Not Found error to return the parent directory
                instead of the file itself, as that is the issue if not present.
-               Until #5792 fixes catching `File_Error.Not_Found` specifically,
-               so instead we catch all `File_Error`s and match the needed one.
             wrapped = stream.catch File_Error error-> case error of
                 File_Error.Not_Found file_path -> Error.throw (File_Error.Not_Found file_path.parent)
                 _ -> stream
             Output_Stream.new wrapped (File_Error.handle_java_exceptions self)
 
         Context.Output.if_enabled disabled_message="File writing is forbidden as the Output context is disabled." panic=False <|
-            Managed_Resource.bracket (new_output_stream self open_options) (_.close) action
+            open_as_data_link = (open_options.contains Data_Link_Access.No_Follow . not) && (Data_Link.is_data_link self)
+            if open_as_data_link then Data_Link.write_data_link_as_stream self open_options action else
+                # We ignore the Data_Link_Access options at this stage:
+                just_file_options = open_options.filter opt-> opt.is_a File_Access
+                Managed_Resource.bracket (new_output_stream self just_file_options) (_.close) action
 
     ## PRIVATE
        Creates a new output stream for this file. Recommended to use
@@ -241,13 +245,18 @@ type File
                  file.with_input_stream [File_Access.Create, File_Access.Read] action
     with_input_stream : Vector File_Access -> (Input_Stream -> Any ! File_Error) -> Any ! File_Error
     with_input_stream self (open_options : Vector) action =
+        new_input_stream : File -> Vector File_Access -> Output_Stream ! File_Error
+        new_input_stream file open_options =
+            opts = open_options . map (_.to_java)
+            stream = File_Error.handle_java_exceptions file (file.input_stream_builtin opts)
+            Input_Stream.new stream (File_Error.handle_java_exceptions self)
+
         if self.is_directory then Error.throw (File_Error.IO_Error self "File '"+self.path+"' is a directory") else
-            new_input_stream : File -> Vector File_Access -> Output_Stream ! File_Error
-            new_input_stream file open_options =
-                opts = open_options . map (_.to_java)
-                stream = File_Error.handle_java_exceptions file (file.input_stream_builtin opts)
-                Input_Stream.new stream (File_Error.handle_java_exceptions self)
-            Managed_Resource.bracket (new_input_stream self open_options) (_.close) action
+            open_as_data_link = (open_options.contains Data_Link_Access.No_Follow . not) && (Data_Link.is_data_link self)
+            if open_as_data_link then Data_Link.read_data_link_as_stream self open_options action else
+                # We ignore the Data_Link_Access options at this stage:
+                just_file_options = open_options.filter opt-> opt.is_a File_Access
+                Managed_Resource.bracket (new_input_stream self just_file_options) (_.close) action
 
     ## ALIAS load, open
        GROUP Input
@@ -284,8 +293,9 @@ type File
     read : File_Format -> Problem_Behavior -> Any ! File_Error
     read self format=Auto_Detect (on_problems=Problem_Behavior.Report_Warning) =
         if self.exists.not then Error.throw (File_Error.Not_Found self) else
-            File_Format.handle_format_missing_arguments format <|
-                format.read self on_problems
+            if Data_Link.is_data_link self then Data_Link.read_data_link self format on_problems else
+                File_Format.handle_format_missing_arguments format <|
+                    format.read self on_problems
 
     ## ALIAS load bytes, open bytes
        ICON data_input

distribution/lib/Standard/Base/0.0.0-dev/src/System/File/Data_Link_Access.enso

@@ -0,0 +1,13 @@
+## PRIVATE
+   ADVANCED
+   Settings regarding how to handle data-links.
+type Data_Link_Access
+    ## PRIVATE
+       The setting that requests from the operation to not follow the data-link,
+       but instead read the raw data-link configuration directly.
+
+       This can be used when working with data-links programmatically.
+
+       If the option is provided for a file that is not a data-link, it is
+       ignored.
+    No_Follow

distribution/lib/Standard/Base/0.0.0-dev/src/System/File/File_Access.enso

@@ -1,3 +1,8 @@
+import project.Data.Text.Text
+import project.Data.Vector.Vector
+import project.Error.Error
+import project.Errors.Illegal_Argument.Illegal_Argument
+
 polyglot java import java.nio.file.StandardOpenOption
 
 ## Represents different options for opening file streams.
@@ -56,3 +61,9 @@ type File_Access
         File_Access.Sync              -> StandardOpenOption.SYNC
         File_Access.Truncate_Existing -> StandardOpenOption.TRUNCATE_EXISTING
         File_Access.Write             -> StandardOpenOption.WRITE
+
+    ## PRIVATE
+    ensure_only_allowed_options (operation_name : Text) (allowed_options : Vector) (got_options : Vector) ~action =
+        disallowed_options = got_options.filter o-> allowed_options.contains o . not
+        if disallowed_options.is_empty then action else
+            Error.throw (Illegal_Argument.Error "Invalid open options for `"+operation_name+"`: "+disallowed_options.to_text+".")

distribution/lib/Standard/Base/0.0.0-dev/src/System/File/Generic/File_Like.enso

@@ -1,5 +1,6 @@
 import project.Data.Text.Text
 import project.System.File.File
+import project.System.File_Format_Metadata.File_Format_Metadata
 
 ## PRIVATE
    A generic interface for file-like objects.
@@ -23,3 +24,8 @@ type File_Like
 
 ## PRIVATE
 File_Like.from (that : Text) = File_Like.from (File.new that)
+
+## PRIVATE
+   If a conversion to `File_Format_Metadata` is needed, we delegate to the underlying file.
+   Every `File_Like` should be able to provide its file format metadata.
+File_Format_Metadata.from (that : File_Like) = File_Format_Metadata.from that.underlying

distribution/lib/Standard/Base/0.0.0-dev/src/System/File/Generic/File_Write_Strategy.enso

@@ -2,6 +2,7 @@ import project.Error.Error
 import project.Errors.File_Error.File_Error
 import project.Panic.Panic
 import project.Runtime.Context
+import project.System.File.Data_Link_Access.Data_Link_Access
 import project.System.File.Existing_File_Behavior.Existing_File_Behavior
 import project.System.File.File
 import project.System.File.File_Access.File_Access
@@ -137,7 +138,8 @@ dry_run_behavior file behavior:Existing_File_Behavior -> Dry_Run_File_Settings =
    Generic `copy` implementation between two backends.
    The files only need to support `with_input_stream` and `with_output_stream`.
 generic_copy source destination replace_existing =
-    source.with_input_stream [File_Access.Read] input_stream->
-        options = if replace_existing then [File_Access.Write, File_Access.Create, File_Access.Truncate_Existing] else [File_Access.Write, File_Access.Create_New]
+    source.with_input_stream [File_Access.Read, Data_Link_Access.No_Follow] input_stream->
+        replace_options = if replace_existing then [File_Access.Create, File_Access.Truncate_Existing] else [File_Access.Create_New]
+        options = [File_Access.Write, Data_Link_Access.No_Follow] + replace_options
         destination.with_output_stream options output_stream->
             output_stream.write_stream input_stream . if_not_error destination

distribution/lib/Standard/Base/0.0.0-dev/src/System/File_Format_Metadata.enso

@@ -27,7 +27,7 @@ type File_Format_Metadata
        - extension: the extension of the file.
        - read_first_bytes: a function that reads the first bytes of the file.
        - content_type: the content type of the file.
-    Value path:Text|Nothing name:Text|Nothing (extension:Text|Nothing = Nothing) (read_first_bytes:(Integer -> Nothing | Vector Integer)=(_->Nothing)) (content_type:Text|Nothing = Nothing)
+    Value (path:Text|Nothing = Nothing) (name:Text|Nothing = Nothing) (extension:Text|Nothing = Nothing) (read_first_bytes:(Integer -> Nothing | Vector Integer)=(_->Nothing)) (content_type:Text|Nothing = Nothing)
 
     ## PRIVATE
        An instance that contains no information at all.
@@ -54,11 +54,18 @@ type File_Format_Metadata
 
     ## PRIVATE
     to_display_text self -> Text =
-        self.path.if_nothing <|
-            self.name.if_nothing <|
-                self.content_type.if_nothing <|
-                    (self.extension.if_not_nothing ("*"+self.extension)) . if_nothing <|
-                        "<Unknown file>"
+        entries = Vector.new_builder
+        self.path.if_not_nothing <|
+            entries.append "path="+self.path
+        self.name.if_not_nothing <|
+            entries.append "name="+self.name
+        self.extension.if_not_nothing <|
+            entries.append "extension="+self.extension
+        self.content_type.if_not_nothing <|
+            entries.append "content_type="+self.content_type
+
+        description = if entries.is_empty then "No information" else entries.to_vector.join ", "
+        "(File_Format_Metadata: "+description+")"
 
 ## PRIVATE
 File_Format_Metadata.from (that:File) = File_Format_Metadata.Value that.path that.name that.extension that.read_first_bytes

distribution/lib/Standard/Base/0.0.0-dev/src/System/Input_Stream.enso

@@ -9,6 +9,7 @@ import project.Runtime.Managed_Resource.Managed_Resource
 import project.System.File.Advanced.Temporary_File.Temporary_File
 import project.System.File.File
 import project.System.File.File_Access.File_Access
+import project.System.File.Generic.Writable_File.Writable_File
 
 polyglot java import java.io.InputStream as Java_Input_Stream
 polyglot java import org.enso.base.encoding.ReportingStreamDecoder
@@ -106,11 +107,10 @@ type Input_Stream
 
     ## PRIVATE
        Reads the contents of this stream into a given file.
-    write_to_file : File -> File
-    write_to_file self file =
+    write_to_file self (file : Writable_File) =
         result = file.with_output_stream [File_Access.Create, File_Access.Truncate_Existing, File_Access.Write] output_stream->
             output_stream.write_stream self
-        result.if_not_error file
+        result.if_not_error file.file
 
 ## PRIVATE
    Utility method for closing primitive Java streams. Provided to avoid

distribution/lib/Standard/Database/0.0.0-dev/src/Connection/Data_Link/Postgres_Data_Link.enso

@@ -1,4 +1,6 @@
 from Standard.Base import all
+import Standard.Base.Errors.Illegal_Argument.Illegal_Argument
+import Standard.Base.System.Input_Stream.Input_Stream
 from Standard.Base.Enso_Cloud.Data_Link import parse_secure_value
 from Standard.Base.Enso_Cloud.Public_Utils import get_optional_field, get_required_field
 
@@ -38,11 +40,12 @@ type Postgres_Data_Link
                 Postgres_Data_Link.Table table_name details
 
     ## PRIVATE
-    read self (on_problems : Problem_Behavior) =
+    read self (format = Auto_Detect) (on_problems : Problem_Behavior) =
         _ = on_problems
-        default_options = Connection_Options.Value
-        connection = self.details.connect default_options
-        case self of
-            Postgres_Data_Link.Connection _ -> connection
-            Postgres_Data_Link.Table table_name _ ->
-                connection.query table_name
+        if format != Auto_Detect then Error.throw (Illegal_Argument.Error "Only the default Auto_Detect format should be used with a Postgres Data Link, because it does not point to a file resource, but a database entity, so setting a file format for it is meaningless.") else
+            default_options = Connection_Options.Value
+            connection = self.details.connect default_options
+            case self of
+                Postgres_Data_Link.Connection _ -> connection
+                Postgres_Data_Link.Table table_name _ ->
+                    connection.query table_name

distribution/lib/Standard/Database/0.0.0-dev/src/DB_Column.enso

@@ -1187,7 +1187,7 @@ type DB_Column
     sort self order=Sort_Direction.Ascending =
         self.to_table.order_by [Sort_Column.Index 0 order] . at 0
 
-    ## ALIAS first, last, sample, slice, top, head, tail, foot, limit
+    ## ALIAS first, last, sample, slice, top, head, tail, limit
        GROUP Standard.Base.Selections
        ICON select_row
        Creates a new Column with the specified range of rows from the input

distribution/lib/Standard/Database/0.0.0-dev/src/DB_Table.enso

@@ -688,7 +688,7 @@ type DB_Table
         result = self.filter column Filter_Condition.Is_True
         Warning.attach (Deprecated.Warning "Standard.Database.DB_Table.DB_Table" "filter_by_expression" "Deprecated: use `filter` with an `Expression` instead.") result
 
-    ## ALIAS first, last, sample, slice, top, head, tail, foot, limit
+    ## ALIAS first, last, sample, slice, top, head, tail, limit
        GROUP Standard.Base.Selections
        ICON select_row
        Creates a new Table with the specified range of rows from the input

distribution/lib/Standard/Examples/0.0.0-dev/src/Main.enso

@@ -153,7 +153,7 @@ get_response = HTTP.fetch geo_data_url . with_materialized_body
      Calling this method will cause Enso to make a network request to a data
      endpoint.
 get_geo_data : Response_Body
-get_geo_data = Data.fetch geo_data_url try_auto_parse_response=False
+get_geo_data = Data.fetch geo_data_url format=Raw_Response
 
 ## A basic URI for examples.
 uri : URI

distribution/lib/Standard/Snowflake/0.0.0-dev/src/Snowflake_Data_Link.enso

@@ -1,4 +1,5 @@
 from Standard.Base import all
+import Standard.Base.Errors.Illegal_Argument.Illegal_Argument
 from Standard.Base.Enso_Cloud.Data_Link import parse_secure_value
 from Standard.Base.Enso_Cloud.Public_Utils import get_optional_field, get_required_field
 
@@ -37,11 +38,12 @@ type Snowflake_Data_Link
                 Snowflake_Data_Link.Table table_name details
 
     ## PRIVATE
-    read self (on_problems : Problem_Behavior) =
+    read self (format = Auto_Detect) (on_problems : Problem_Behavior) =
         _ = on_problems
-        default_options = Connection_Options.Value
-        connection = self.details.connect default_options
-        case self of
-            Snowflake_Data_Link.Connection _ -> connection
-            Snowflake_Data_Link.Table table_name _ ->
-                connection.query table_name
+        if format != Auto_Detect then Error.throw (Illegal_Argument.Error "Only the default Auto_Detect format should be used with a Snowflake Data Link, because it does not point to a file resource, but a database entity, so setting a file format for it is meaningless.") else
+            default_options = Connection_Options.Value
+            connection = self.details.connect default_options
+            case self of
+                Snowflake_Data_Link.Connection _ -> connection
+                Snowflake_Data_Link.Table table_name _ ->
+                    connection.query table_name

distribution/lib/Standard/Table/0.0.0-dev/src/Data/Column.enso

@@ -2290,7 +2290,7 @@ type Column
             sorted = self.to_vector.sort order by=wrapped
             Column.from_vector self.name sorted
 
-    ## ALIAS first, last, sample, slice, top, head, tail, foot, limit
+    ## ALIAS first, last, sample, slice, top, head, tail, limit
        GROUP Standard.Base.Selections
        ICON select_row
        Creates a new Column with the specified range of rows from the input

distribution/lib/Standard/Table/0.0.0-dev/src/Data/Table.enso

@@ -1526,7 +1526,7 @@ type Table
         result = self.filter column Filter_Condition.Is_True
         Warning.attach (Deprecated.Warning "Standard.Table.Data.Table.Table" "filter_by_expression" "Deprecated: use `filter` with an `Expression` instead.") result
 
-    ## ALIAS first, last, sample, slice, top, head, tail, foot, limit
+    ## ALIAS first, last, sample, slice, top, head, tail, limit
        GROUP Standard.Base.Selections
        ICON select_row
        Creates a new Table with the specified range of rows from the input

std-bits/base/src/main/java/org/enso/base/enso_cloud/DataLinkFileFormatSPI.java

@@ -1,17 +0,0 @@
-package org.enso.base.enso_cloud;
-
-import org.enso.base.file_format.FileFormatSPI;
-
-/** A format registration for parsing `.datalink` files as data links. */
-@org.openide.util.lookup.ServiceProvider(service = FileFormatSPI.class)
-public class DataLinkFileFormatSPI extends FileFormatSPI {
-  @Override
-  protected String getModuleName() {
-    return "Standard.Base.Enso_Cloud.Data_Link";
-  }
-
-  @Override
-  protected String getTypeName() {
-    return "Data_Link_Format";
-  }
-}

test/AWS_Tests/src/Inter_Backend_File_Operations_Spec.enso

@@ -0,0 +1,158 @@
+## This test file checks operations on files that are happening between various backends.
+
+   Because it relies not only on Standard.Base but also the S3 backend provided
+   by Standard.AWS, it is currently placed in `AWS_Tests`.
+   Once we start supporting more backends, we should consider creating
+   a separate test project for these integrations (e.g. `Integrator_Tests`).
+
+from Standard.Base import all
+import Standard.Base.Enso_Cloud.Data_Link.Data_Link_Format
+import Standard.Base.Errors.File_Error.File_Error
+import Standard.Base.Errors.Illegal_Argument.Illegal_Argument
+
+from Standard.AWS import S3_File
+
+from Standard.Test import all
+
+from project.S3_Spec import api_pending, writable_root
+
+add_specs suite_builder =
+    my_writable_s3_dir = writable_root / "inter-backend-test-run-"+(Date_Time.now.format "yyyy-MM-dd_HHmmss.fV" . replace "/" "|")+"/"
+    sources = [my_writable_s3_dir / "source1.txt", File.create_temporary_file "source2" ".txt"]
+    destinations = [my_writable_s3_dir / "destination1.txt", File.create_temporary_file "destination2" ".txt"]
+    sources.each source_file-> destinations.each destination_file-> if source_file.is_a File && destination_file.is_a File then Nothing else
+        src_typ = Meta.type_of source_file . to_display_text
+        dest_typ = Meta.type_of destination_file . to_display_text
+        suite_builder.group "("+src_typ+" -> "+dest_typ+") copying/moving" pending=api_pending group_builder->
+            group_builder.teardown <|
+                source_file.delete_if_exists
+                destination_file.delete_if_exists
+
+            group_builder.specify "should be able to copy files" <|
+                "Hello".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
+                destination_file.delete_if_exists
+
+                source_file.copy_to destination_file . should_succeed
+                destination_file.read . should_equal "Hello"
+                source_file.exists . should_be_true
+
+            group_builder.specify "should be able to move files" <|
+                "Hello".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
+                destination_file.delete_if_exists
+
+                source_file.move_to destination_file . should_succeed
+                destination_file.read . should_equal "Hello"
+                source_file.exists . should_be_false
+
+            group_builder.specify "should fail if the source file does not exist" <|
+                source_file.delete_if_exists
+                destination_file.delete_if_exists
+
+                r = source_file.copy_to destination_file
+                r.should_fail_with File_Error
+                r.catch.should_be_a File_Error.Not_Found
+
+                r2 = source_file.move_to destination_file
+                r2.should_fail_with File_Error
+                r2.catch.should_be_a File_Error.Not_Found
+
+                destination_file.exists . should_be_false
+
+            group_builder.specify "should fail to copy/move a file if it exists and replace_existing=False" <|
+                "Hello".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
+                "World".write destination_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
+
+                r = source_file.copy_to destination_file
+                r.should_fail_with File_Error
+                r.catch.should_be_a File_Error.Already_Exists
+
+                r2 = source_file.move_to destination_file
+                r2.should_fail_with File_Error
+                r2.catch.should_be_a File_Error.Already_Exists
+
+                destination_file.read . should_equal "World"
+
+            group_builder.specify "should overwrite existing destination in copy/move if replace_existing=True" <|
+                "Hello".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
+                "World".write destination_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
+
+                source_file.copy_to destination_file replace_existing=True . should_succeed
+                destination_file.read . should_equal "Hello"
+                source_file.exists . should_be_true
+
+                "FooBar".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
+                source_file.move_to destination_file replace_existing=True . should_succeed
+                destination_file.read . should_equal "FooBar"
+                source_file.exists . should_be_false
+
+    sample_data_link_content = Data_Link_Format.read_raw_config (enso_project.data / "simple.datalink")
+    # TODO Enso_File datalink once Enso_File & cloud datalink write is supported
+    data_link_sources = [my_writable_s3_dir / "s3.datalink", File.create_temporary_file "local" ".datalink"]
+    data_link_destinations = [my_writable_s3_dir / "s3-target.datalink", File.create_temporary_file "local-target" ".datalink"]
+    ## This introduces a lot of combinations for testing the datalink copy/move logic, but unfortunately it is needed,
+       because various combinations of backends may rely on different logic (different operations happen under the hood
+       if a file is moved locally vs if it is moved from a local filesystem to S3 or vice versa), and all that different
+       logic may be prone to mis-handling datalinks - so we need to test all paths to ensure coverage.
+    data_link_sources.each source_file-> data_link_destinations.each destination_file->
+        src_typ = Meta.type_of source_file . to_display_text
+        dest_typ = Meta.type_of destination_file . to_display_text
+        suite_builder.group "("+src_typ+" -> "+dest_typ+") Data Link copying/moving" pending=api_pending group_builder->
+            group_builder.teardown <|
+                source_file.delete_if_exists
+                destination_file.delete_if_exists
+
+            group_builder.specify "should be able to copy a datalink file to a regular file" <|
+                regular_destination_file = destination_file.parent / destination_file.name+".txt"
+                regular_destination_file.delete_if_exists
+                Data_Link_Format.write_raw_config source_file sample_data_link_content replace_existing=True . should_succeed
+
+                source_file.copy_to regular_destination_file . should_succeed
+                Panic.with_finalizer regular_destination_file.delete_if_exists <|
+                    # The raw datalink config is copied, so reading back the .txt file yields us the configuration:
+                    regular_destination_file.read . should_contain '"libraryName": "Standard.AWS"'
+
+            group_builder.specify "should be able to copy a datalink file to another datalink" <|
+                destination_file.delete_if_exists
+                Data_Link_Format.write_raw_config source_file sample_data_link_content replace_existing=True . should_succeed
+
+                source_file.copy_to destination_file replace_existing=True . should_succeed
+                # Now the destination is _also_ a datalink, pointing to the same target as source, so reading it yields the target data:
+                destination_file.read . should_equal "Hello WORLD!"
+
+                # But if we read it raw, we can see that it is still a datalink, not just a copy of the data:
+                Data_Link_Format.read_raw_config destination_file . should_equal sample_data_link_content
+
+            group_builder.specify "should be able to move a datalink" <|
+                destination_file.delete_if_exists
+                Data_Link_Format.write_raw_config source_file sample_data_link_content replace_existing=True . should_succeed
+
+                source_file.move_to destination_file . should_succeed
+                source_file.exists . should_be_false
+
+                destination_file.read . should_equal "Hello WORLD!"
+                Data_Link_Format.read_raw_config destination_file . should_equal sample_data_link_content
+
+            group_builder.specify "should be able to move a regular file to become a datalink" <|
+                destination_file.delete_if_exists
+
+                regular_source_file = source_file.parent / source_file.name+".txt"
+                sample_data_link_content.write regular_source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
+                Panic.with_finalizer regular_source_file.delete_if_exists <|
+                    # The source file is not a datalink - it is a raw text file, so reading it gives us back the raw config
+                    regular_source_file.read . should_equal sample_data_link_content
+                    # Reading it raw will even fail:
+                    Test.expect_panic Illegal_Argument <|
+                        Data_Link_Format.read_raw_config regular_source_file
+
+                    regular_source_file.move_to destination_file . should_succeed
+                    regular_source_file.exists . should_be_false
+
+                    # However, the destination file _is_ a datalink, so it is read as target data:
+                    destination_file.read . should_equal "Hello WORLD!"
+                    # Unless we read it raw:
+                    Data_Link_Format.read_raw_config destination_file . should_equal sample_data_link_content
+
+main filter=Nothing =
+    suite = Test.build suite_builder->
+        add_specs suite_builder
+    suite.run_with_filter filter

test/AWS_Tests/src/Main.enso

@@ -2,9 +2,11 @@ from Standard.Base import all
 
 from Standard.Test import all
 
+import project.Inter_Backend_File_Operations_Spec
 import project.S3_Spec
 
 main filter=Nothing =
     suite = Test.build suite_builder->
         S3_Spec.add_specs suite_builder
+        Inter_Backend_File_Operations_Spec.add_specs suite_builder
     suite.run_with_filter filter

test/AWS_Tests/src/S3_Spec.enso

@@ -1,17 +1,17 @@
 from Standard.Base import all
 from Standard.Base.Runtime import assert
+import Standard.Base.Enso_Cloud.Data_Link.Data_Link_Format
 import Standard.Base.Errors.Common.Forbidden_Operation
 import Standard.Base.Errors.File_Error.File_Error
 import Standard.Base.Errors.Illegal_Argument.Illegal_Argument
 import Standard.Base.Runtime.Context
-import Standard.Base.Runtime.Ref.Ref
 
 from Standard.AWS import S3, S3_File, AWS_Credential
 from Standard.AWS.Errors import AWS_SDK_Error, More_Records_Available, S3_Error, S3_Bucket_Not_Found, S3_Key_Not_Found
 import Standard.AWS.Internal.S3_Path.S3_Path
 
 # Needed for custom formats test
-from Standard.Table import Table, Excel_Format
+from Standard.Table import Table, Excel_Format, Delimited
 # Needed for correct `Table.should_equal`
 import enso_dev.Table_Tests.Util
 
@@ -20,19 +20,40 @@ from Standard.Test import all
 import enso_dev.Base_Tests.Network.Enso_Cloud.Cloud_Tests_Setup.Cloud_Tests_Setup
 from enso_dev.Base_Tests.Network.Enso_Cloud.Cloud_Tests_Setup import with_retries
 
+
+api_pending = if Environment.get "AWS_ACCESS_KEY_ID" . is_nothing then "No Access Key found." else Nothing
+
+bucket_name = "enso-data-samples"
+writable_bucket_name = "enso-ci-s3-test-stage"
+writable_root = S3_File.new "s3://"+writable_bucket_name+"/"
+not_a_bucket_name = "not_a_bucket_enso"
+object_name = "Bus_Stop_Benches.geojson"
+folder_name = "examples/"
+sub_folder_name = "examples/folder 1/"
+root = S3_File.new "s3://"+bucket_name+"/"
+hello_txt = S3_File.new "s3://"+bucket_name+"/examples/folder 2/hello.txt"
+
+delete_on_panic file ~action =
+    handler caught_panic =
+        file.delete
+        Panic.throw caught_panic
+    Panic.catch Any action handler
+
+delete_afterwards file ~action =
+    Panic.with_finalizer file.delete action
+
+
+## Reads the datalink as plain text and replaces the placeholder username with
+   actual one. It then writes the new contents to a temporary file and returns
+   it.
+replace_username_in_data_link base_file =
+    content = Data_Link_Format.read_raw_config base_file
+    new_content = content.replace "USERNAME" Enso_User.current.name
+    temp_file = File.create_temporary_file prefix=base_file.name suffix=base_file.extension
+    Data_Link_Format.write_raw_config temp_file new_content replace_existing=True . if_not_error temp_file
+
 add_specs suite_builder =
-    bucket_name = "enso-data-samples"
-    writable_bucket_name = "enso-ci-s3-test-stage"
-    not_a_bucket_name = "not_a_bucket_enso"
-    object_name = "Bus_Stop_Benches.geojson"
-    folder_name = "examples/"
-    sub_folder_name = "examples/folder 1/"
-    api_pending = if Environment.get "AWS_ACCESS_KEY_ID" . is_nothing then "No Access Key found." else Nothing
     cloud_setup = Cloud_Tests_Setup.prepare
-
-    root = S3_File.new "s3://"+bucket_name+"/"
-    hello_txt = S3_File.new "s3://"+bucket_name+"/examples/folder 2/hello.txt"
-
     suite_builder.group "S3 Path handling" group_builder->
         group_builder.specify "parse bucket only uris" <|
             S3_Path.parse "s3://" . should_equal (S3_Path.Value "" "")
@@ -259,15 +280,7 @@ add_specs suite_builder =
             # AWS S3 does not record creation time, only last modified time.
             hello_txt.creation_time . should_fail_with S3_Error
 
-    writable_root = S3_File.new "s3://"+writable_bucket_name+"/"
     my_writable_dir = writable_root / "test-run-"+(Date_Time.now.format "yyyy-MM-dd_HHmmss.fV" . replace "/" "|")+"/"
-    delete_on_panic file ~action =
-        handler caught_panic =
-            file.delete
-            Panic.throw caught_panic
-        Panic.catch Any action handler
-    delete_afterwards file ~action =
-        Panic.with_finalizer file.delete action
 
     suite_builder.group "S3_File writing" pending=api_pending group_builder->
         assert my_writable_dir.is_directory
@@ -338,7 +351,8 @@ add_specs suite_builder =
             r.catch.to_display_text . should_contain "read it, modify and then write the new contents"
 
             r2 = new_file.with_output_stream [File_Access.Read] _->Nothing
-            r2.should_fail_with S3_Error
+            r2.should_fail_with Illegal_Argument
+            r2.catch.to_display_text . should_contain "Invalid open options for `with_output_stream`"
 
         group_builder.specify "will respect the File_Access.Create_New option and fail if the file already exists" <|
             new_file = my_writable_dir / "new_file-stream-create-new.txt"
@@ -477,80 +491,13 @@ add_specs suite_builder =
             r.should_fail_with File_Error
             r.catch.should_be_a File_Error.Not_Found
 
-    sources = [my_writable_dir / "source1.txt", File.create_temporary_file "source2" ".txt"]
-    destinations = [my_writable_dir / "destination1.txt", File.create_temporary_file "destination2" ".txt"]
-    sources.each source_file-> destinations.each destination_file-> if source_file.is_a File && destination_file.is_a File then Nothing else
-        src_typ = Meta.type_of source_file . to_display_text
-        dest_typ = Meta.type_of destination_file . to_display_text
-        suite_builder.group "("+src_typ+" -> "+dest_typ+") copying/moving" pending=api_pending group_builder->
-            group_builder.teardown <|
-                source_file.delete_if_exists
-                destination_file.delete_if_exists
-
-            group_builder.specify "should be able to copy files" <|
-                "Hello".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
-                destination_file.delete_if_exists
-
-                source_file.copy_to destination_file . should_succeed
-                destination_file.read . should_equal "Hello"
-                source_file.exists . should_be_true
-
-            group_builder.specify "should be able to move files" <|
-                "Hello".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
-                destination_file.delete_if_exists
-
-                source_file.move_to destination_file . should_succeed
-                destination_file.read . should_equal "Hello"
-                source_file.exists . should_be_false
-
-            group_builder.specify "should fail if the source file does not exist" <|
-                source_file.delete_if_exists
-                destination_file.delete_if_exists
-
-                r = source_file.copy_to destination_file
-                r.should_fail_with File_Error
-                r.catch.should_be_a File_Error.Not_Found
-
-                r2 = source_file.move_to destination_file
-                r2.should_fail_with File_Error
-                r2.catch.should_be_a File_Error.Not_Found
-
-                destination_file.exists . should_be_false
-
-            group_builder.specify "should fail to copy/move a file if it exists and replace_existing=False" <|
-                "Hello".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
-                "World".write destination_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
-
-                r = source_file.copy_to destination_file
-                r.should_fail_with File_Error
-                r.catch.should_be_a File_Error.Already_Exists
-
-                r2 = source_file.move_to destination_file
-                r2.should_fail_with File_Error
-                r2.catch.should_be_a File_Error.Already_Exists
-
-                destination_file.read . should_equal "World"
-
-            group_builder.specify "should overwrite existing destination in copy/move if replace_existing=True" <|
-                "Hello".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
-                "World".write destination_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
-
-                source_file.copy_to destination_file replace_existing=True . should_succeed
-                destination_file.read . should_equal "Hello"
-                source_file.exists . should_be_true
-
-                "FooBar".write source_file on_existing_file=Existing_File_Behavior.Overwrite . should_succeed
-                source_file.move_to destination_file replace_existing=True . should_succeed
-                destination_file.read . should_equal "FooBar"
-                source_file.exists . should_be_false
-
     suite_builder.group "DataLinks to S3_File" pending=api_pending group_builder->
         group_builder.specify "should be able to read a data link of an S3 File" <|
             # It reads the datalink description and then reads the actual S3 file contents:
             (enso_project.data / "simple.datalink") . read . should_equal "Hello WORLD!"
 
         group_builder.specify "should be able to read a data link with custom credentials and secrets" pending=cloud_setup.pending <| cloud_setup.with_prepared_environment <|
-            transformed_datalink_file = replace_username_in_datalink (enso_project.data / "credentials-with-secrets.datalink")
+            transformed_data_link_file = replace_username_in_data_link (enso_project.data / "credentials-with-secrets.datalink")
 
             secret_key_id = Enso_Secret.create "datalink-secret-AWS-keyid" (Environment.get "AWS_ACCESS_KEY_ID")
             secret_key_id.should_succeed
@@ -558,7 +505,7 @@ add_specs suite_builder =
                 secret_key_value = Enso_Secret.create "datalink-secret-AWS-secretkey" (Environment.get "AWS_SECRET_ACCESS_KEY")
                 secret_key_value.should_succeed
                 Panic.with_finalizer secret_key_value.delete <| with_retries <|
-                    transformed_datalink_file.read . should_equal "Hello WORLD!"
+                    transformed_data_link_file.read . should_equal "Hello WORLD!"
 
         group_builder.specify "should be able to read a data link with a custom file format set" <|
             r = (enso_project.data / "format-delimited.datalink") . read
@@ -567,16 +514,24 @@ add_specs suite_builder =
             r.column_names . should_equal ["Column 1", "Column 2"]
             r.rows.at 0 . to_vector . should_equal ["Hello", "WORLD!"]
 
+        group_builder.specify "should be able to read a data link stored on S3" <|
+            s3_link = my_writable_dir / "my-simple.datalink"
+            raw_content = Data_Link_Format.read_raw_config (enso_project.data / "simple.datalink")
+            Data_Link_Format.write_raw_config s3_link raw_content replace_existing=True . should_succeed
+            Panic.with_finalizer s3_link.delete <|
+                s3_link.read . should_equal "Hello WORLD!"
+
+        group_builder.specify "should be able to read an S3 data link overriding the format" <|
+            s3_link = my_writable_dir / "my-simple.datalink"
+            raw_content = Data_Link_Format.read_raw_config (enso_project.data / "simple.datalink")
+            Data_Link_Format.write_raw_config s3_link raw_content replace_existing=True . should_succeed
+            Panic.with_finalizer s3_link.delete <|
+                r = s3_link.read (Delimited " " headers=False)
+                r.should_be_a Table
+                r.column_names . should_equal ["Column 1", "Column 2"]
+                r.rows.at 0 . to_vector . should_equal ["Hello", "WORLD!"]
+
 main filter=Nothing =
     suite = Test.build suite_builder->
         add_specs suite_builder
     suite.run_with_filter filter
-
-## Reads the datalink as plain text and replaces the placeholder username with
-   actual one. It then writes the new contents to a temporary file and returns
-   it.
-replace_username_in_datalink base_file =
-    content = base_file.read Plain_Text
-    new_content = content.replace "USERNAME" Enso_User.current.name
-    temp_file = File.create_temporary_file prefix=base_file.name suffix=base_file.extension
-    new_content.write temp_file . if_not_error temp_file

test/Base_Tests/src/Network/Enso_Cloud/Cloud_Data_Link_Spec.enso

@@ -1,5 +1,8 @@
 from Standard.Base import all
+import Standard.Base.Enso_Cloud.Data_Link.Data_Link_Format
+import Standard.Base.Enso_Cloud.Errors.Missing_Data_Link_Library
 import Standard.Base.Errors.Common.Not_Found
+import Standard.Base.Errors.Illegal_Argument.Illegal_Argument
 import Standard.Base.Errors.Illegal_State.Illegal_State
 import Standard.Base.Enso_Cloud.Enso_File.Enso_Asset_Type
 
@@ -31,9 +34,16 @@ add_specs suite_builder setup:Cloud_Tests_Setup = setup.with_prepared_environmen
             datalink.asset_type.should_equal Enso_Asset_Type.Data_Link
 
             r = datalink.read
-            r.should_fail_with Illegal_State
+            r.should_fail_with Missing_Data_Link_Library
             r.catch.to_display_text . should_contain "The data link for S3 is provided by the library Standard.AWS which is not loaded."
 
+        group_builder.specify "does not allow to use Data_Link_Format to read/write regular files" <|
+            temp_file = File.create_temporary_file "not-a-datalink" ".txt"
+            Test.expect_panic Illegal_Argument <|
+                Data_Link_Format.write_raw_config temp_file "{}"
+            Test.expect_panic Illegal_Argument <|
+                Data_Link_Format.read_raw_config temp_file
+
 main filter=Nothing =
     setup = Cloud_Tests_Setup.prepare
     suite = Test.build suite_builder->

test/Base_Tests/src/Network/Http/Http_Auto_Parse_Spec.enso

@@ -32,19 +32,19 @@ add_specs suite_builder =
             url_windows_1250.fetch . should_equal content_windows_1250
 
         group_builder.specify "should detect the encoding from Content-Type in decode_as_text" <|
-            r1 = url_utf8.fetch try_auto_parse_response=False
+            r1 = url_utf8.fetch format=Raw_Response
             r1.decode_as_text . should_equal content_utf
 
-            r2 = url_windows_1250.fetch try_auto_parse_response=False
+            r2 = url_windows_1250.fetch format=Raw_Response
             r2.decode_as_text . should_equal content_windows_1250
 
-            r3 = url_utf8.fetch try_auto_parse_response=False
+            r3 = url_utf8.fetch format=Raw_Response
             # We may override the encoding detected from Content-Type:
             r3.decode_as_text Encoding.ascii . should_fail_with Encoding_Error
 
         group_builder.specify "should detect the encoding from Content-Type in decode_as_json" <|
-            r1 = url_utf8.fetch try_auto_parse_response=False
+            r1 = url_utf8.fetch format=Raw_Response
             r1.decode_as_json . should_equal ["x", "Hello! 😊👍 ąę"]
 
-            r2 = url_windows_1250.fetch try_auto_parse_response=False
+            r2 = url_windows_1250.fetch format=Raw_Response
             r2.decode_as_json . should_equal ["y", "Hello! ąę"]

test/Base_Tests/src/Network/Http/Http_Data_Link_Spec.enso

@@ -1,5 +1,8 @@
 from Standard.Base import all
+import Standard.Base.Enso_Cloud.Data_Link.Data_Link_Format
+import Standard.Base.Errors.File_Error.File_Error
 import Standard.Base.Errors.Illegal_State.Illegal_State
+import Standard.Base.Network.HTTP.Response.Response
 
 from Standard.Test import all
 
@@ -15,40 +18,99 @@ add_specs suite_builder =
     data_link_root = enso_project.data / "datalinks"
     suite_builder.group "HTTP DataLink" pending=pending_has_url group_builder->
         group_builder.specify "should allow to read a web resource" <|
-            f = replace_url_in_datalink (data_link_root / "example-http.datalink")
+            f = replace_url_in_data_link (data_link_root / "example-http.datalink")
             r = f.read
             # Defaults to reading as text, because the resource read is called `js.txt`, implying Plain_Text format
             r.should_be_a Text
             r.trim.should_equal '{"hello": "world"}'
 
         group_builder.specify "should allow to read a web resource, with explicitly using default format" <|
-            f = replace_url_in_datalink (data_link_root / "example-http-format-explicit-default.datalink")
+            f = replace_url_in_data_link (data_link_root / "example-http-format-explicit-default.datalink")
             r = f.read
             r.should_be_a Text
             r.trim.should_equal '{"hello": "world"}'
 
         group_builder.specify "should allow to read a web resource, setting format to JSON" <|
-            f = replace_url_in_datalink (data_link_root / "example-http-format-json.datalink")
+            f = replace_url_in_data_link (data_link_root / "example-http-format-json.datalink")
             r = f.read
             js = '{"hello": "world"}'.parse_json
             r.should_equal js
             r.get "hello" . should_equal "world"
 
         group_builder.specify "will fail if invalid format is used" <|
-            r1 = (data_link_root / "example-http-format-invalid.datalink").read
-            r1.should_fail_with Illegal_State
-            r1.catch.to_display_text.should_contain "Unknown format"
+            f = replace_url_in_data_link (data_link_root / "example-http-format-invalid.datalink")
+            r = f.read
+            r.should_fail_with Illegal_State
+            r.catch.to_display_text.should_contain "Unknown format"
 
+        group_builder.specify "will fail if an unloaded format is used" <|
             # We assume that Base_Tests _do not_ import Standard.Table
-            r2 = (data_link_root / "example-http-format-delimited.datalink").read
-            r2.should_fail_with Illegal_State
-            r2.catch.to_display_text.should_contain "Unknown format"
+            f = replace_url_in_data_link (data_link_root / "example-http-format-delimited.datalink")
+            r = f.read
+            r.should_fail_with Illegal_State
+            r.catch.to_display_text.should_contain "Unknown format"
+
+        group_builder.specify "but will succeed if an unknown format is not used because it was overridden" <|
+            f = replace_url_in_data_link (data_link_root / "example-http-format-delimited.datalink")
+            r = f.read Plain_Text
+            r.should_be_a Text
+            r.trim.should_equal '{"hello": "world"}'
+
+        group_builder.specify "should be able to follow a datalink from HTTP in Data.read" <|
+            r1 = Data.read base_url_with_slash+"dynamic.datalink" JSON_Format
+            r1.should_equal ('{"hello": "world"}'.parse_json)
+
+            r2 = Data.read base_url_with_slash+"dynamic-datalink" Plain_Text
+            r2.trim.should_equal '{"hello": "world"}'
+
+        group_builder.specify "should be able to follow a datalink from HTTP in Data.fetch/post, if auto parse is on" <|
+            r1 = Data.fetch base_url_with_slash+"dynamic.datalink"
+            r1.trim.should_equal '{"hello": "world"}'
+
+            r2 = Data.fetch base_url_with_slash+"dynamic-datalink"
+            r2.trim.should_equal '{"hello": "world"}'
+
+            r3 = Data.post base_url_with_slash+"dynamic.datalink"
+            r3.trim.should_equal '{"hello": "world"}'
+
+        group_builder.specify "will return raw datalink config data in Data.fetch/post if auto parse is off" <|
+            r1 = Data.fetch base_url_with_slash+"dynamic.datalink" format=Raw_Response
+            r1.should_be_a Response
+
+            ## Normally .datalink is not a valid format, so we cannot decode it,
+               assuming that the server did not add some content type that would
+               make us treat the file e.g. as a text file.
+            r1_decoded = r1.decode
+            r1_decoded.should_fail_with File_Error
+            r1_decoded.catch . should_be_a File_Error.Unsupported_Type
+
+            # Still raw data link config is returned if we successfully decode it by overriding the format.
+            r1_plain = r1.decode Plain_Text
+            r1_plain.should_contain '"libraryName": "Standard.Base"'
+
+            r2 = Data.post base_url_with_slash+"dynamic-datalink" response_format=Raw_Response
+            r2.should_be_a Response
+            r2_decoded = r2.decode
+            r2_decoded.should_fail_with File_Error
+            r2_decoded.catch . should_be_a File_Error.Unsupported_Type
+
+            r2_plain = r2.decode Plain_Text
+            r2_plain.should_contain '"libraryName": "Standard.Base"'
+
+        group_builder.specify "should follow a datalink encountered in Data.download" <|
+            target_file = enso_project.data / "transient" / "my_download.txt"
+            target_file.delete_if_exists
+            Data.download base_url_with_slash+"dynamic.datalink" target_file . should_equal target_file
+
+            Panic.with_finalizer target_file.delete_if_exists <|
+                target_file.read.trim.should_equal '{"hello": "world"}'
+
 
 ## Reads the datalink as plain text and replaces the placeholder URL with actual
    URL of the server. It then writes the new contents to a temporary file and
    returns it.
-replace_url_in_datalink base_file =
-    content = base_file.read Plain_Text
+replace_url_in_data_link base_file =
+    content = Data_Link_Format.read_raw_config base_file
     new_content = content.replace "http://http-test-helper.local/" base_url_with_slash
     temp_file = File.create_temporary_file prefix=base_file.name suffix=base_file.extension
-    new_content.write temp_file . if_not_error temp_file
+    Data_Link_Format.write_raw_config temp_file new_content replace_existing=True . if_not_error temp_file

test/Base_Tests/src/Network/Http_Spec.enso

@@ -71,6 +71,11 @@ add_specs suite_builder =
             http = HTTP.new (follow_redirects = False)
             http.follow_redirects.should_equal False
 
+            r = http.request (Request.new HTTP_Method.Get base_url_with_slash+"test_redirect")
+            r.should_fail_with HTTP_Error
+            r.catch.should_be_a HTTP_Error.Status_Error
+            r.catch.status_code.code . should_equal 302
+
         group_builder.specify "should create HTTP client with proxy setting" <|
             proxy_setting = Proxy.Address "example.com" 80
             http = HTTP.new (proxy = proxy_setting)
@@ -81,8 +86,8 @@ add_specs suite_builder =
             http = HTTP.new (version = version_setting)
             http.version.should_equal version_setting
 
+    url_get = base_url_with_slash + "get"
     suite_builder.group "fetch" pending=pending_has_url group_builder->
-        url_get = base_url_with_slash + "get"
         url_head = base_url_with_slash + "head"
         url_options = base_url_with_slash + "options"
 
@@ -133,7 +138,7 @@ add_specs suite_builder =
             uri_response.at "headers" . at "Content-Length" . should_equal "0"
 
         group_builder.specify "Can skip auto-parse" <|
-            response = Data.fetch url_get try_auto_parse_response=False
+            response = Data.fetch url_get format=Raw_Response
             response.code.code . should_equal 200
             expected_response = Json.parse <| '''
                 {
@@ -151,15 +156,15 @@ add_specs suite_builder =
                 }
             compare_responses response.decode_as_json expected_response
 
-            uri_response = url_get.to_uri.fetch try_auto_parse_response=False
+            uri_response = url_get.to_uri.fetch format=Raw_Response
             uri_response.code.code . should_equal 200
             compare_responses uri_response.decode_as_json expected_response
 
         group_builder.specify "Can still perform request when output context is disabled" <|
             run_with_and_without_output <|
-                Data.fetch url_get try_auto_parse_response=False . code . code . should_equal 200
-                Data.fetch url_get method=HTTP_Method.Head try_auto_parse_response=False . code . code . should_equal 200
-                Data.fetch url_get method=HTTP_Method.Options try_auto_parse_response=False . code . code . should_equal 200
+                Data.fetch url_get format=Raw_Response . code . code . should_equal 200
+                Data.fetch url_get method=HTTP_Method.Head format=Raw_Response . code . code . should_equal 200
+                Data.fetch url_get method=HTTP_Method.Options format=Raw_Response . code . code . should_equal 200
 
         group_builder.specify "Unsupported method" <|
             err = Data.fetch url_get method=HTTP_Method.Post
@@ -188,6 +193,27 @@ add_specs suite_builder =
             header_names.should_not_contain "http2-settings"
             header_names.should_not_contain "upgrade"
 
+    suite_builder.group "HTTP in Data.read" pending=pending_has_url group_builder->
+        group_builder.specify "can use URI in Data.read" <|
+            r = Data.read (URI.from url_get)
+            r.should_be_a JS_Object
+
+        group_builder.specify "works if HTTP is uppercase" <|
+            r = Data.fetch (url_get.replace "http" "HTTP")
+            r.should_be_a JS_Object
+
+        group_builder.specify "should follow redirects" <|
+            r = Data.read base_url_with_slash+"test_redirect"
+            r.should_be_a Text
+            r.trim . should_equal '{"hello": "world"}'
+
+        group_builder.specify "can override the format" <|
+            auto_response = Data.read url_get
+            auto_response.should_be_a JS_Object
+
+            plain_response = Data.read url_get format=Plain_Text
+            plain_response.should_be_a Text
+
     suite_builder.group "post" pending=pending_has_url group_builder->
         url_post = base_url_with_slash + "post"
         url_put = base_url_with_slash + "put"
@@ -318,7 +344,7 @@ add_specs suite_builder =
             compare_responses response expected_response
 
         group_builder.specify "Can skip auto-parse" <|
-            response = Data.post url_post (Request_Body.Text "hello world") try_auto_parse_response=False
+            response = Data.post url_post (Request_Body.Text "hello world") response_format=Raw_Response
             expected_response = echo_response_template "POST" "/post" "hello world" content_type="text/plain; charset=UTF-8"
             compare_responses response.decode_as_json expected_response
 
@@ -501,7 +527,7 @@ add_specs suite_builder =
             Data.post url_post (Request_Body.Text "hello world" encoding=Encoding.utf_8) headers=[Header.content_type "application/json"] . should_fail_with Illegal_Argument
 
         group_builder.specify "can also read headers from a response, when returning a raw response" <|
-            r1 = Data.post url_post (Request_Body.Text "hello world") try_auto_parse_response=False
+            r1 = Data.post url_post (Request_Body.Text "hello world") response_format=Raw_Response
             r1.should_be_a Response
             # The result is JSON data:
             r1.headers.find (p-> p.name.equals_ignore_case "Content-Type") . value . should_equal "application/json"
@@ -514,7 +540,7 @@ add_specs suite_builder =
             uri = URI.from (base_url_with_slash + "test_headers")
                 . add_query_argument "test-header" "test-value"
                 . add_query_argument "Other-Header" "some other value"
-            r2 = Data.fetch uri try_auto_parse_response=False
+            r2 = Data.fetch uri format=Raw_Response
             r2.should_be_a Response
             r2.headers.find (p-> p.name.equals_ignore_case "Test-Header") . value . should_equal "test-value"
             r2.headers.find (p-> p.name.equals_ignore_case "Other-Header") . value . should_equal "some other value"
@@ -524,7 +550,7 @@ add_specs suite_builder =
                 . add_query_argument "my-header" "value-1"
                 . add_query_argument "my-header" "value-2"
                 . add_query_argument "my-header" "value-44"
-            r1 = Data.fetch uri try_auto_parse_response=False
+            r1 = Data.fetch uri format=Raw_Response
             r1.should_be_a Response
             my_headers = r1.headers.filter (p-> p.name.equals_ignore_case "my-header") . map .value
             my_headers.sort . should_equal ["value-1", "value-2", "value-44"]

test/Table_Tests/src/Database/Postgres_Spec.enso

@@ -1,7 +1,9 @@
 from Standard.Base import all
+import Standard.Base.Enso_Cloud.Data_Link.Data_Link_Format
 import Standard.Base.Errors.Illegal_Argument.Illegal_Argument
 import Standard.Base.Errors.Illegal_State.Illegal_State
 import Standard.Base.Runtime.Ref.Ref
+import Standard.Base.System.File.Data_Link_Access.Data_Link_Access
 
 from Standard.Table import Table, Value_Type, Aggregate_Column, Bits, expr
 from Standard.Table.Errors import Invalid_Column_Names, Inexact_Type_Coercion, Duplicate_Output_Column_Names
@@ -28,6 +30,7 @@ from project.Common_Table_Operations.Util import all
 from project.Database.Types.Postgres_Type_Mapping_Spec import default_text
 
 import enso_dev.Base_Tests.Network.Enso_Cloud.Cloud_Tests_Setup.Cloud_Tests_Setup
+import enso_dev.Base_Tests.Network.Http.Http_Test_Setup
 from enso_dev.Base_Tests.Network.Enso_Cloud.Cloud_Tests_Setup import with_retries
 
 
@@ -860,37 +863,37 @@ add_connection_setup_specs suite_builder = suite_builder.group "[PostgreSQL] Con
                     [c2, c3, c4].each c->
                         c.jdbc_properties . should_equal <| add_ssl [Pair.new "user" "other user", Pair.new "password" "other password"]
 
-add_datalink_specs suite_builder =
+add_data_link_specs suite_builder =
     connection_details = get_configured_connection_details
     pending = if connection_details.is_nothing then "PostgreSQL test database is not configured. See README.md for instructions."
 
     # This transforms the datalink file, replacing prepared constants with actual values.
     transform_file base_file =
-        contents = base_file.read Plain_Text
-        new_contents = contents
+        content = Data_Link_Format.read_raw_config base_file
+        new_content = content
             . replace "HOSTNAME" connection_details.host
             . replace "12345" connection_details.port.to_text
             . replace "DBNAME" connection_details.database
             . replace "USERNAME" connection_details.credentials.username
             . replace "PASSWORD" connection_details.credentials.password
-        tmp_file = File.create_temporary_file base_file.name base_file.extension
-        new_contents.write tmp_file . if_not_error tmp_file
+        temp_file = File.create_temporary_file base_file.name base_file.extension
+        Data_Link_Format.write_raw_config temp_file new_content replace_existing=True . if_not_error temp_file
 
     suite_builder.group "[PostgreSQL] DataLink" pending=pending group_builder->
         group_builder.specify "should be able to open a datalink setting up a connection to the database" <|
-            datalink_file = transform_file (enso_project.data / "datalinks" / "postgres-db.datalink")
+            data_link_file = transform_file (enso_project.data / "datalinks" / "postgres-db.datalink")
 
-            datalink_connection = Data.read datalink_file
-            Panic.with_finalizer datalink_connection.close <|
-                datalink_connection.tables.column_names . should_contain "Name"
+            data_link_connection = Data.read data_link_file
+            Panic.with_finalizer data_link_connection.close <|
+                data_link_connection.tables.column_names . should_contain "Name"
 
                 # Test that this is really a DB connection:
-                q = datalink_connection.query 'SELECT 1 AS "A"'
+                q = data_link_connection.query 'SELECT 1 AS "A"'
                 q.column_names . should_equal ["A"]
                 q.at "A" . to_vector . should_equal [1]
 
         group_builder.specify "should be able to open a datalink to a particular database table" <|
-            datalink_file = transform_file (enso_project.data / "datalinks" / "postgres-table.datalink")
+            data_link_file = transform_file (enso_project.data / "datalinks" / "postgres-table.datalink")
             connection = Database.connect connection_details
             Panic.with_finalizer connection.close <|
                 # We create the table that will then be accessed through the datalink, and ensure it's cleaned up afterwards.
@@ -899,17 +902,40 @@ add_datalink_specs suite_builder =
                 Panic.with_finalizer (connection.drop_table example_table.name) <|
                     ## Now we access this table but this time through a datalink.
                        Btw. this will keep a connection open until the table is garbage collected, but that is probably fine...
-                    datalink_table = Data.read datalink_file
-                    datalink_table.should_be_a DB_Table
-                    datalink_table.column_names . should_equal ["X", "Y"]
-                    datalink_table.at "X" . to_vector . should_equal [22]
-                    datalink_table.at "Y" . to_vector . should_equal ["o"]
+                    data_link_table = Data.read data_link_file
+                    data_link_table.should_be_a DB_Table
+                    data_link_table.column_names . should_equal ["X", "Y"]
+                    data_link_table.at "X" . to_vector . should_equal [22]
+                    data_link_table.at "Y" . to_vector . should_equal ["o"]
+
+        group_builder.specify "will reject any format overrides or stream operations on the data link" <|
+            data_link_file = transform_file (enso_project.data / "datalinks" / "postgres-db.datalink")
+
+            r1 = Data.read data_link_file Plain_Text
+            r1.should_fail_with Illegal_Argument
+            r1.catch.to_display_text . should_contain "Only the default Auto_Detect format should be used"
+
+            r2 = data_link_file.with_input_stream [File_Access.Read] .read_all_bytes
+            r2.should_fail_with Illegal_Argument
+
+            # But we can read the raw data link if we ask for it:
+            r3 = data_link_file.with_input_stream [File_Access.Read, Data_Link_Access.No_Follow] .read_all_bytes
+            r3.should_be_a Vector
+
+        group_builder.specify "will fail with a clear message if trying to download a Database datalink" pending=Http_Test_Setup.pending_has_url <|
+            url = Http_Test_Setup.base_url_with_slash+"testfiles/some-postgres.datalink"
+            target_file = enso_project.data / "transient" / "some-postgres-target"
+            target_file.delete_if_exists
+            r = Data.download url target_file
+            r.should_fail_with Illegal_Argument
+            r.catch.to_display_text . should_contain "The Postgres Data Link cannot be saved to a file."
+
 
 add_specs suite_builder =
     add_table_specs suite_builder
     add_pgpass_specs suite_builder
     add_connection_setup_specs suite_builder
-    add_datalink_specs suite_builder
+    add_data_link_specs suite_builder
 
 main filter=Nothing =
     suite = Test.build suite_builder->

test/Table_Tests/src/IO/Data_Link_Formats_Spec.enso

@@ -1,11 +1,13 @@
 from Standard.Base import all
+import Standard.Base.System.File.Data_Link_Access.Data_Link_Access
+import Standard.Base.System.File_Format_Metadata.File_Format_Metadata
 
 from Standard.Table import all
 
 from Standard.Test import all
 
 from enso_dev.Base_Tests.Network.Http.Http_Test_Setup import base_url_with_slash, pending_has_url
-from enso_dev.Base_Tests.Network.Http.Http_Data_Link_Spec import replace_url_in_datalink
+from enso_dev.Base_Tests.Network.Http.Http_Data_Link_Spec import replace_url_in_data_link
 
 import project.Util
 
@@ -18,33 +20,65 @@ main filter=Nothing =
 add_specs suite_builder = suite_builder.group "parsing Table formats in DataLinks" pending=pending_has_url group_builder->
     data_link_root = enso_project.data / "datalinks"
     group_builder.specify "parsing Delimited without quotes" <|
-        datalink_file = replace_url_in_datalink (data_link_root / "example-http-format-delimited-ignore-quote.datalink")
-        t = datalink_file.read
+        data_link_file = replace_url_in_data_link (data_link_root / "example-http-format-delimited-ignore-quote.datalink")
+        t = data_link_file.read
         t.should_equal (Table.from_rows ["Column 1", "Column 2"] [['{"hello":', '"world"}']])
 
     group_builder.specify "parsing Delimited with custom delimiter quotes" <|
-        datalink_file = replace_url_in_datalink (data_link_root / "example-http-format-delimited-custom-quote.datalink")
-        t = datalink_file.read
+        data_link_file = replace_url_in_data_link (data_link_root / "example-http-format-delimited-custom-quote.datalink")
+        t = data_link_file.read
         weird_txt = "x'z" + '""w'
         # The A column remains a text column because of being quoted
         t.should_equal (Table.new [["A", ["1", "3"]], ["B", [weird_txt, "y"]]])
 
-    group_builder.specify "parsing Excel_Format.Workbook" <|
-        datalink_file = replace_url_in_datalink (data_link_root / "example-http-format-excel-workbook.datalink")
+    group_builder.specify "overriding the custom format in Delimited datalink" <|
+        data_link_file = replace_url_in_data_link (data_link_root / "example-http-format-delimited-ignore-quote.datalink")
+        r = data_link_file.read Plain_Text
+        r.should_be_a Text
+        r.trim.should_equal '{"hello": "world"}'
 
-        workbook = datalink_file.read
+    group_builder.specify "parsing Excel_Format.Workbook" <|
+        data_link_file = replace_url_in_data_link (data_link_root / "example-http-format-excel-workbook.datalink")
+
+        workbook = data_link_file.read
         Panic.with_finalizer workbook.close <|
             workbook.should_be_a Excel_Workbook
             workbook.sheet_names . should_equal ["MyTestSheet"]
 
     group_builder.specify "parsing Excel_Format.Sheet" <|
-        datalink_file = replace_url_in_datalink (data_link_root / "example-http-format-excel-sheet.datalink")
+        data_link_file = replace_url_in_data_link (data_link_root / "example-http-format-excel-sheet.datalink")
 
-        table = datalink_file.read
+        table = data_link_file.read
         table . should_equal (Table.from_rows ["A", "B"] [[1, 'x'], [3, 'y']])
 
     group_builder.specify "parsing Excel_Format.Range" <|
-        datalink_file = replace_url_in_datalink (data_link_root / "example-http-format-excel-range.datalink")
+        data_link_file = replace_url_in_data_link (data_link_root / "example-http-format-excel-range.datalink")
 
-        table = datalink_file.read
+        table = data_link_file.read
         table . should_equal (Table.from_rows ["A", "B"] [[3, 'y']])
+
+    group_builder.specify "overriding Excel format" <|
+        data_link_file = replace_url_in_data_link (data_link_root / "example-http-format-excel-workbook.datalink")
+
+        table = data_link_file.read (Excel_Format.Range "MyTestSheet!A1:B1")
+        table . should_equal (Table.from_rows ["A", "B"] [[1, 'x']])
+
+        bytes = data_link_file.read_bytes
+        bytes.should_be_a Vector
+
+    group_builder.specify "reading a datalink as a stream" <|
+        data_link_file = replace_url_in_data_link (data_link_root / "example-http-format-excel-range.datalink")
+
+        r1 = data_link_file.with_input_stream [File_Access.Read] input_stream->
+            ## We need to specify the format explicitly because the raw stream
+               has no access to file metadata and the excel reader needs to know if its XLS or XLSX.
+            metadata = File_Format_Metadata.Value extension=".xls"
+            Excel_Format.Workbook.read_stream input_stream metadata
+        r1.should_be_a Excel_Workbook
+
+        r2 = data_link_file.with_input_stream [File_Access.Read] .read_all_bytes
+        r2.should_be_a Vector
+
+        # But it is still possible to access the raw data too.
+        r3 = data_link_file.with_input_stream [File_Access.Read, Data_Link_Access.No_Follow] .read_all_bytes
+        Text.from_bytes r3 Encoding.utf_8 . should_contain '"type": "HTTP"'

test/Table_Tests/src/IO/Fetch_Spec.enso

@@ -47,7 +47,7 @@ add_specs suite_builder =
             r.sheet_names . should_equal ["MyTestSheet"]
             r.read "MyTestSheet" . should_equal expected_table
 
-            r2 = Data.fetch url try_auto_parse_response=False . decode (Excel_Format.Sheet "MyTestSheet")
+            r2 = Data.fetch url format=Raw_Response . decode (Excel_Format.Sheet "MyTestSheet")
             r2.should_be_a Table
             r2.should_equal expected_table
 
@@ -60,11 +60,11 @@ add_specs suite_builder =
             r.sheet_names . should_equal ["MyTestSheet"]
             r.read "MyTestSheet" . should_equal expected_table
 
-            r2 = Data.fetch url try_auto_parse_response=False . decode (Excel_Format.Sheet "MyTestSheet")
+            r2 = Data.fetch url format=Raw_Response . decode (Excel_Format.Sheet "MyTestSheet")
             r2.should_be_a Table
             r2.should_equal expected_table
 
-            r3 = url.to_uri.fetch try_auto_parse_response=False . decode (Excel_Format.Sheet "MyTestSheet")
+            r3 = url.to_uri.fetch format=Raw_Response . decode (Excel_Format.Sheet "MyTestSheet")
             r3.should_be_a Table
             r3.should_equal expected_table
 

tools/http-test-helper/src/main/java/org/enso/shttp/HTTPTestHelperServer.java

@@ -14,6 +14,7 @@ import org.enso.shttp.auth.TokenAuthTestHandler;
 import org.enso.shttp.cloud_mock.CloudAuthRenew;
 import org.enso.shttp.cloud_mock.CloudRoot;
 import org.enso.shttp.cloud_mock.ExpiredTokensCounter;
+import org.enso.shttp.test_helpers.*;
 import sun.misc.Signal;
 import sun.misc.SignalHandler;
 
@@ -81,16 +82,24 @@ public class HTTPTestHelperServer {
       server.addHandler(path, new TestHandler(method));
     }
 
+    // HTTP helpers
+    setupFileServer(server, projectRoot);
     server.addHandler("/test_headers", new HeaderTestHandler());
     server.addHandler("/test_token_auth", new TokenAuthTestHandler());
     server.addHandler("/test_basic_auth", new BasicAuthTestHandler());
     server.addHandler("/crash", new CrashingTestHandler());
+    server.addHandler("/test_redirect", new RedirectTestHandler("/testfiles/js.txt"));
+
+    // Cloud mock
     var expiredTokensCounter = new ExpiredTokensCounter();
     server.addHandler("/COUNT-EXPIRED-TOKEN-FAILURES", expiredTokensCounter);
     CloudRoot cloudRoot = new CloudRoot(expiredTokensCounter);
     server.addHandler(cloudRoot.prefix, cloudRoot);
     server.addHandler("/enso-cloud-auth-renew", new CloudAuthRenew());
-    setupFileServer(server, projectRoot);
+
+    // Data link helpers
+    server.addHandler("/dynamic-datalink", new GenerateDataLinkHandler(true));
+    server.addHandler("/dynamic.datalink", new GenerateDataLinkHandler(false));
   }
 
   private static void setupFileServer(HybridHTTPServer server, Path projectRoot) {

tools/http-test-helper/src/main/java/org/enso/shttp/SimpleHttpHandler.java

@@ -32,8 +32,15 @@ public abstract class SimpleHttpHandler implements HttpHandler {
 
   protected final void sendResponse(int code, String message, HttpExchange exchange)
       throws IOException {
+    sendResponse(code, message, exchange, "text/plain; charset=utf-8");
+  }
+
+  protected final void sendResponse(
+      int code, String message, HttpExchange exchange, String contentType) throws IOException {
     byte[] response = message.getBytes(StandardCharsets.UTF_8);
-    exchange.getResponseHeaders().add("Content-Type", "text/plain; charset=utf-8");
+    if (contentType != null) {
+      exchange.getResponseHeaders().add("Content-Type", contentType);
+    }
     exchange.sendResponseHeaders(code, response.length);
     try (OutputStream os = exchange.getResponseBody()) {
       os.write(response);

tools/http-test-helper/src/main/java/org/enso/shttp/test_helpers/CrashingTestHandler.java

@@ -1,7 +1,8 @@
-package org.enso.shttp;
+package org.enso.shttp.test_helpers;
 
 import com.sun.net.httpserver.HttpExchange;
 import java.io.IOException;
+import org.enso.shttp.SimpleHttpHandler;
 
 public class CrashingTestHandler extends SimpleHttpHandler {
   @Override

tools/http-test-helper/src/main/java/org/enso/shttp/test_helpers/GenerateDataLinkHandler.java

@@ -0,0 +1,33 @@
+package org.enso.shttp.test_helpers;
+
+import com.sun.net.httpserver.HttpExchange;
+import java.io.IOException;
+import org.enso.shttp.SimpleHttpHandler;
+
+/** A handler that generates a Data Link pointing to a file on this server. */
+public class GenerateDataLinkHandler extends SimpleHttpHandler {
+  private final boolean includeContentType;
+  private static final String targetPath = "/testfiles/js.txt";
+  private static final String dataLinkTemplate =
+      """
+      {
+          "type": "HTTP",
+          "libraryName": "Standard.Base",
+          "method": "GET",
+          "uri": "${URI}"
+      }
+      """;
+
+  public GenerateDataLinkHandler(boolean includeContentType) {
+    this.includeContentType = includeContentType;
+  }
+
+  @Override
+  protected void doHandle(HttpExchange exchange) throws IOException {
+    String host = exchange.getRequestHeaders().getFirst("Host");
+    String uri = "http://" + host + targetPath;
+    String content = dataLinkTemplate.replace("${URI}", uri);
+    String contentType = includeContentType ? "application/x-enso-datalink" : null;
+    sendResponse(200, content, exchange, contentType);
+  }
+}

tools/http-test-helper/src/main/java/org/enso/shttp/test_helpers/HeaderTestHandler.java

@@ -1,10 +1,11 @@
-package org.enso.shttp;
+package org.enso.shttp.test_helpers;
 
 import com.sun.net.httpserver.HttpExchange;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.net.URI;
 import org.apache.http.client.utils.URIBuilder;
+import org.enso.shttp.SimpleHttpHandler;
 
 public class HeaderTestHandler extends SimpleHttpHandler {
   @Override

tools/http-test-helper/src/main/java/org/enso/shttp/test_helpers/RedirectTestHandler.java

@@ -0,0 +1,19 @@
+package org.enso.shttp.test_helpers;
+
+import com.sun.net.httpserver.HttpExchange;
+import java.io.IOException;
+import org.enso.shttp.SimpleHttpHandler;
+
+public class RedirectTestHandler extends SimpleHttpHandler {
+  private final String redirectLocation;
+
+  public RedirectTestHandler(String redirectLocation) {
+    this.redirectLocation = redirectLocation;
+  }
+
+  @Override
+  protected void doHandle(HttpExchange exchange) throws IOException {
+    exchange.getResponseHeaders().add("Location", redirectLocation);
+    exchange.sendResponseHeaders(302, -1);
+  }
+}

tools/http-test-helper/src/main/java/org/enso/shttp/test_helpers/TestHandler.java

@@ -1,4 +1,4 @@
-package org.enso.shttp;
+package org.enso.shttp.test_helpers;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -14,6 +14,8 @@ import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import org.apache.http.NameValuePair;
 import org.apache.http.client.utils.URIBuilder;
+import org.enso.shttp.HttpMethod;
+import org.enso.shttp.SimpleHttpHandler;
 
 public class TestHandler extends SimpleHttpHandler {
   private final HttpMethod expectedMethod;

tools/http-test-helper/www-files/some-postgres.datalink

@@ -0,0 +1,11 @@
+{
+  "type": "Postgres_Connection",
+  "libraryName": "Standard.Database",
+  "host": "example.com",
+  "port": 12345,
+  "database_name": "DBNAME",
+  "credentials": {
+    "username": "USERNAME",
+    "password": "PASSWORD"
+  }
+}