Skip to content

Commit

Permalink
feat(*): add namespace as a parameter of the internal storage
Browse files Browse the repository at this point in the history
  • Loading branch information
@loicmathieu
loicmathieu committed Nov 21, 2024
1 parent 94c9a3c commit cc9116d
Show file tree
Hide file tree
Showing 2 changed files with 20 additions and 17 deletions.
2 changes: 1 addition & 1 deletion gradle.properties
View file
Original file line number Diff line number Diff line change
@@ -1,2 +1,2 @@
version=0.20.0-SNAPSHOT
kestraVersion=[0.18,)
kestraVersion=[0.20,)
35 changes: 19 additions & 16 deletions src/main/java/io/kestra/storage/gcs/GcsStorage.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;

import static io.kestra.core.utils.Rethrow.throwFunction;

@AllArgsConstructor
Expand Down Expand Up @@ -104,12 +106,12 @@ private void parentTraversalGuard(URI uri) {
}

@Override
public InputStream get(String tenantId, URI uri) throws IOException {
return getWithMetadata(tenantId, uri).inputStream();
public InputStream get(String tenantId, @Nullable String namespace, URI uri) throws IOException {
return getWithMetadata(tenantId, namespace, uri).inputStream();
}

@Override
public StorageObject getWithMetadata(String tenantId, URI uri) throws IOException {
public StorageObject getWithMetadata(String tenantId, @Nullable String namespace, URI uri) throws IOException {
try {
Blob blob = this.storage.get(this.blob(tenantId, URI.create(uri.getPath())));

Expand All @@ -125,7 +127,7 @@ public StorageObject getWithMetadata(String tenantId, URI uri) throws IOExceptio
}

@Override
public List<URI> allByPrefix(String tenantId, URI prefix, boolean includeDirectories) {
public List<URI> allByPrefix(String tenantId, @Nullable String namespace, URI prefix, boolean includeDirectories) {
String path = getPath(tenantId, prefix);
return blobsForPrefix(path, true, includeDirectories)
.map(BlobInfo::getName)
Expand All @@ -134,7 +136,7 @@ public List<URI> allByPrefix(String tenantId, URI prefix, boolean includeDirecto
}

@Override
public List<FileAttributes> list(String tenantId, URI uri) throws IOException {
public List<FileAttributes> list(String tenantId, @Nullable String namespace, URI uri) throws IOException {
String path = getPath(tenantId, uri);
String prefix = (path.endsWith("/")) ? path : path + "/";

Expand All @@ -143,7 +145,7 @@ public List<FileAttributes> list(String tenantId, URI uri) throws IOException {
.toList();
if(list.isEmpty()) {
// this will throw FileNotFound if there is no directory
this.getAttributes(tenantId, uri);
this.getAttributes(tenantId, namespace, uri);
}
return list;
}
Expand All @@ -167,7 +169,7 @@ private Stream<Blob> blobsForPrefix(String prefix, boolean recursive, boolean in
}

@Override
public boolean exists(String tenantId, URI uri) {
public boolean exists(String tenantId, @Nullable String namespace, URI uri) {
try {
Blob blob = this.storage.get(this.blob(tenantId, URI.create(uri.getPath())));
return blob != null && blob.exists();
Expand All @@ -177,9 +179,9 @@ public boolean exists(String tenantId, URI uri) {
}

@Override
public FileAttributes getAttributes(String tenantId, URI uri) throws IOException {
public FileAttributes getAttributes(String tenantId, @Nullable String namespace, URI uri) throws IOException {
String path = getPath(tenantId, uri);
if (!exists(tenantId, uri)) {
if (!exists(tenantId, namespace, uri)) {
path = path + "/";
}
Blob blob = this.storage.get(this.blob(path));
Expand All @@ -200,7 +202,7 @@ private FileAttributes getGcsFileAttributes(Blob blob) {
}

@Override
public URI put(String tenantId, URI uri, StorageObject storageObject) throws IOException {
public URI put(String tenantId, @Nullable String namespace, URI uri, StorageObject storageObject) throws IOException {
try {
String path = getPath(tenantId, uri);
mkdirs(path);
Expand Down Expand Up @@ -255,10 +257,10 @@ private void mkdirs(String path) {
}

@Override
public boolean delete(String tenantId, URI uri) throws IOException {
public boolean delete(String tenantId, @Nullable String namespace, URI uri) throws IOException {
FileAttributes fileAttributes;
try {
fileAttributes = getAttributes(tenantId, uri);
fileAttributes = getAttributes(tenantId, namespace, uri);
} catch (FileNotFoundException e) {
return false;
}
Expand All @@ -267,6 +269,7 @@ public boolean delete(String tenantId, URI uri) throws IOException {
if (fileAttributes.getType() == FileAttributes.FileType.Directory) {
return !this.deleteByPrefix(
tenantId,
namespace,
uri.getPath().endsWith("/") ? uri : URI.create(uri.getPath() + "/")
).isEmpty();
}
Expand All @@ -275,7 +278,7 @@ public boolean delete(String tenantId, URI uri) throws IOException {
}

@Override
public URI createDirectory(String tenantId, URI uri) {
public URI createDirectory(String tenantId, @Nullable String namespace, URI uri) {
String path = getPath(tenantId, uri);
if (!path.endsWith("/")) {
path = path + "/";
Expand All @@ -285,11 +288,11 @@ public URI createDirectory(String tenantId, URI uri) {
}

@Override
public URI move(String tenantId, URI from, URI to) throws IOException {
public URI move(String tenantId, @Nullable String namespace, URI from, URI to) throws IOException {
String path = getPath(tenantId, from);
StorageBatch batch = this.storage.batch();

if (getAttributes(tenantId, from).getType() == FileAttributes.FileType.File) {
if (getAttributes(tenantId, namespace, from).getType() == FileAttributes.FileType.File) {
// move just a file
BlobId source = blob(path);
BlobId target = blob(tenantId, to);
Expand All @@ -314,7 +317,7 @@ private void moveFile(BlobId source, BlobId target, StorageBatch batch) {
}

@Override
public List<URI> deleteByPrefix(String tenantId, URI storagePrefix) throws IOException {
public List<URI> deleteByPrefix(String tenantId, @Nullable String namespace, URI storagePrefix) throws IOException {
try {
StorageBatch batch = this.storage.batch();
Map<URI, StorageBatchResult<Boolean>> results = new HashMap<>();
Expand Down

0 comments on commit cc9116d

Please sign in to comment.