Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F81766118
onedrivefs.py
No One
Temporary
Actions
Download File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Subscribers
None
File Metadata
Details
File Info
Storage
Attached
Created
Sat, Sep 7, 20:07
Size
29 KB
Mime Type
text/x-python
Expires
Mon, Sep 9, 20:07 (1 d, 23 h)
Engine
blob
Format
Raw Data
Handle
20617060
Attached To
R3600 invenio-infoscience
onedrivefs.py
View Options
# -*- coding: utf-8 -*-
#
# This file is part of Invenio.
# Copyright (C) 2013, 2014 CERN.
#
# Invenio is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation; either version 2 of the
# License, or (at your option) any later version.
#
# Invenio is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Invenio; if not, write to the Free Software Foundation, Inc.,
# 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
"""
OneDrive file system
--------------------
Known issues:
* Flush and close, both call write contents and because of that
the file on cloud is overwrite twice...
"""
from
onedrive
import
api_v5
import
six
import
os
import
time
from
UserDict
import
UserDict
# python filesystem imports
from
fs.base
import
FS
from
fs.errors
import
(
UnsupportedError
,
CreateFailedError
,
ResourceInvalidError
,
ResourceNotFoundError
,
NoPathURLError
,
OperationFailedError
,
RemoteConnectionError
)
from
fs.remote
import
RemoteFileBuffer
from
fs.filelike
import
SpooledTemporaryFile
# Items in cache are considered expired after 5 minutes.
CACHE_TTL
=
300
# Max size for spooling to memory before using disk (5M).
MAX_BUFFER
=
1024
**
2
*
5
class
CacheItem
(
object
):
"""Represents a path in the cache. There are two components to a path.
It's individual metadata, and the children contained within it."""
def
__init__
(
self
,
metadata
=
None
,
children
=
None
,
timestamp
=
None
,
parent
=
None
):
self
.
metadata
=
metadata
self
.
children
=
children
self
.
parent
=
parent
if
timestamp
is
None
:
timestamp
=
time
.
time
()
self
.
timestamp
=
timestamp
def
add_child
(
self
,
name
,
client
=
None
):
if
self
.
children
is
None
:
if
client
is
not
None
:
# This is a fix. When you add a child to a folder that
# was still not listed, that folder gets only one
# child when you list it afterwards. So this fix
# first tries to check are the files/folders inside
# this directory on cloud.
client
.
children
(
self
.
metadata
[
'id'
])
else
:
self
.
children
=
[
name
]
else
:
if
name
not
in
self
.
children
:
self
.
children
.
append
(
name
)
def
del_child
(
self
,
name
):
if
self
.
children
is
None
:
return
try
:
i
=
self
.
children
.
index
(
name
)
except
ValueError
:
return
self
.
children
.
pop
(
i
)
def
_get_expired
(
self
):
if
self
.
timestamp
<=
time
.
time
()
-
CACHE_TTL
:
return
True
expired
=
property
(
_get_expired
)
def
renew
(
self
):
self
.
timestamp
=
time
.
time
()
class
OneDriveCache
(
UserDict
):
def
__init__
(
self
,
client
):
self
.
_client
=
client
UserDict
.
__init__
(
self
)
def
set
(
self
,
path
,
metadata
,
children
=
None
,
parent
=
None
):
self
[
path
]
=
CacheItem
(
metadata
,
children
=
children
,
parent
=
parent
)
if
parent
is
not
None
:
if
parent
in
self
:
self
.
get
(
parent
)
.
add_child
(
path
,
self
.
_client
)
def
pop
(
self
,
path
,
default
=
None
):
value
=
UserDict
.
pop
(
self
,
path
,
default
)
if
value
.
parent
is
not
None
:
if
value
.
parent
in
self
:
self
.
get
(
value
.
parent
)
.
del_child
(
value
.
metadata
[
'id'
])
return
value
class
OneDriveClient
(
api_v5
.
OneDriveAPI
):
def
__init__
(
self
,
credentials
):
self
.
cache
=
OneDriveCache
(
self
)
self
.
auth_access_token
=
credentials
.
get
(
"access_token"
,
None
)
self
.
auth_refresh_token
=
credentials
.
get
(
"refresh_token"
,
None
)
self
.
auth_redirect_uri
=
credentials
.
get
(
"redirect_uri"
,
None
)
self
.
auth_scope
=
credentials
.
get
(
"scope"
,
None
)
self
.
client_id
=
credentials
.
get
(
"client_id"
,
None
)
self
.
client_secret
=
credentials
.
get
(
"client_secret"
,
None
)
def
metadata
(
self
,
path
):
"Gets metadata for a given path."
item
=
self
.
cache
.
get
(
path
)
if
not
item
or
item
.
metadata
is
None
or
item
.
expired
:
try
:
metadata
=
super
(
OneDriveClient
,
self
)
.
info
(
path
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
code
==
404
:
raise
ResourceNotFoundError
(
path
)
raise
OperationFailedError
(
opname
=
'metadata'
,
path
=
path
,
msg
=
str
(
e
))
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
item
=
self
.
cache
[
path
]
=
CacheItem
(
metadata
)
# Copy the info so the caller cannot affect our cache.
return
dict
(
item
.
metadata
.
items
())
def
children
(
self
,
path
):
"Gets children of a given path."
update
=
False
item
=
self
.
cache
.
get
(
path
)
if
item
:
if
item
.
expired
:
update
=
True
else
:
if
item
.
metadata
[
"type"
]
!=
"folder"
and
\
not
(
"folder"
in
path
):
raise
ResourceInvalidError
(
path
)
if
not
item
.
children
:
update
=
True
else
:
update
=
True
if
update
:
try
:
metadata
=
super
(
OneDriveClient
,
self
)
.
info
(
path
)
if
metadata
[
"type"
]
!=
"folder"
and
not
(
"folder"
in
path
):
raise
ResourceInvalidError
(
path
)
children
=
[]
contents
=
super
(
OneDriveClient
,
self
)
.
listdir
(
path
)
for
child
in
contents
:
children
.
append
(
child
[
'id'
])
self
.
cache
[
child
[
'id'
]]
=
CacheItem
(
child
,
parent
=
path
)
item
=
self
.
cache
[
path
]
=
CacheItem
(
metadata
,
children
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
code
==
404
:
raise
ResourceNotFoundError
(
path
)
if
not
item
or
e
.
resp
.
status
!=
304
:
raise
OperationFailedError
(
opname
=
'metadata'
,
path
=
path
,
msg
=
str
(
e
))
# We have an item from cache (perhaps expired), but it's
# hash is still valid (as far as OneDrive is concerned),
# so just renew it and keep using it.
item
.
renew
()
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
return
item
.
children
def
file_create_folder
(
self
,
parent_id
,
title
):
"Add newly created directory to cache."
try
:
metadata
=
super
(
OneDriveClient
,
self
)
.
mkdir
(
title
,
parent_id
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
code
==
405
:
raise
ResourceInvalidError
(
parent_id
)
if
e
.
code
==
404
:
raise
ResourceNotFoundError
(
parent_id
)
raise
OperationFailedError
(
opname
=
'file_create_folder'
,
msg
=
str
(
e
))
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
self
.
cache
.
set
(
metadata
[
"id"
],
metadata
,
parent
=
parent_id
)
return
metadata
[
'id'
]
def
file_copy
(
self
,
src
,
dst
):
try
:
metadata
=
super
(
OneDriveClient
,
self
)
.
copy
(
src
,
dst
,
False
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
code
==
404
:
raise
ResourceNotFoundError
(
"Parent or source file don't exist"
)
raise
OperationFailedError
(
opname
=
'file_copy'
,
msg
=
str
(
e
))
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
self
.
cache
.
set
(
metadata
[
'id'
],
metadata
,
parent
=
dst
)
return
metadata
[
'id'
]
def
file_move
(
self
,
src
,
dst
):
try
:
metadata
=
super
(
OneDriveClient
,
self
)
.
copy
(
src
,
dst
,
True
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
code
==
404
:
raise
ResourceNotFoundError
(
"Parent or source file don't exist"
)
raise
OperationFailedError
(
opname
=
'file_move'
,
msg
=
str
(
e
))
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
self
.
cache
.
pop
(
src
,
None
)
self
.
cache
.
set
(
metadata
[
'id'
],
metadata
,
metadata
[
'parent_id'
])
return
metadata
[
'id'
]
def
file_delete
(
self
,
path
):
try
:
super
(
OneDriveClient
,
self
)
.
delete
(
path
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
code
==
404
:
raise
ResourceNotFoundError
(
path
)
raise
OperationFailedError
(
opname
=
'file_delete'
,
msg
=
str
(
e
))
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
self
.
cache
.
pop
(
path
,
None
)
def
put_file
(
self
,
parent_id
,
title
,
content
,
overwrite
=
False
):
try
:
metadata
=
super
(
OneDriveClient
,
self
)
.
put
(
(
title
,
content
),
parent_id
,
overwrite
=
overwrite
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
code
==
404
:
raise
ResourceNotFoundError
(
parent_id
)
raise
OperationFailedError
(
opname
=
'put_copy'
,
msg
=
str
(
e
))
except
TypeError
as
e
:
raise
ResourceInvalidError
(
"put_file"
)
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
return
metadata
[
'id'
]
def
get_file
(
self
,
path
):
metadata
=
None
if
not
self
.
cache
.
get
(
path
,
None
):
try
:
metadata
=
super
(
OneDriveClient
,
self
)
.
info
(
path
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
code
==
404
:
raise
ResourceNotFoundError
(
"Source file doesn't exist"
)
raise
OperationFailedError
(
opname
=
'get_file'
,
msg
=
str
(
e
))
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
self
.
cache
.
set
(
metadata
[
'id'
],
metadata
,
parent
=
metadata
[
'parent_id'
])
else
:
item
=
self
.
cache
[
path
]
metadata
=
item
.
metadata
return
super
(
OneDriveClient
,
self
)
.
get
(
path
)
def
update_file
(
self
,
file_id
,
new_file_info
):
try
:
metadata
=
super
(
OneDriveClient
,
self
)
.
info_update
(
file_id
,
new_file_info
)
except
api_v5
.
ProtocolError
as
e
:
if
e
.
resp
.
status
==
404
:
raise
ResourceNotFoundError
(
path
=
file_id
)
raise
OperationFailedError
(
opname
=
'update_file'
,
msg
=
e
.
resp
.
reason
)
except
:
raise
RemoteConnectionError
(
"Most probable reasons: access token has expired "
"or user credentials are invalid."
)
self
.
cache
.
pop
(
file_id
,
None
)
self
.
cache
.
set
(
metadata
[
'id'
],
metadata
,
parent
=
metadata
[
'parent_id'
])
return
metadata
[
'id'
]
class
OneDriveFS
(
FS
):
"""
Sky drive file system
"""
__name__
=
"OneDrive"
_meta
=
{
'thread_safe'
:
True
,
'virtual'
:
False
,
'read_only'
:
False
,
'unicode_paths'
:
True
,
'case_insensitive_paths'
:
False
,
'network'
:
True
,
'atomic.move'
:
True
,
'atomic.copy'
:
True
,
'atomic.makedir'
:
True
,
'atomic.rename'
:
False
,
'atomic.setconetns'
:
True
}
def
__init__
(
self
,
root
=
None
,
credentials
=
None
,
thread_synchronize
=
True
):
self
.
_root
=
root
self
.
_credentials
=
credentials
self
.
cached_files
=
{}
if
self
.
_root
is
None
or
self
.
_root
==
"/"
or
self
.
_root
==
""
:
self
.
_root
=
"me/skydrive"
if
self
.
_credentials
is
None
:
if
"ONEDRIVE_ACCESS_TOKEN"
not
in
os
.
environ
:
raise
CreateFailedError
(
"ONEDRIVE_ACCESS_TOKEN is not set in os.environ"
)
else
:
self
.
_credentials
[
'access_token'
]
=
os
.
environ
.
get
(
'ONEDRIVE_ACCESS_TOKEN'
)
self
.
_credentials
[
'refresh_token'
]
=
os
.
environ
.
get
(
'ONEDRIVE_REFRESH_TOKEN'
,
None
)
self
.
_credentials
[
'redirect_uri'
]
=
os
.
environ
.
get
(
'ONEDRIVE_REDIRECT_URI'
,
None
)
self
.
_credentials
[
'client_id'
]
=
os
.
environ
.
get
(
'ONEDRIVE_CLIENT_ID'
,
None
)
self
.
_credentials
[
'client_secret'
]
=
os
.
environ
.
get
(
'ONEDRIVE_CLIENT_SECRET'
,
None
)
self
.
client
=
OneDriveClient
(
self
.
_credentials
)
super
(
OneDriveFS
,
self
)
.
__init__
(
thread_synchronize
=
thread_synchronize
)
def
__repr__
(
self
):
args
=
(
self
.
__class__
.
__name__
,
self
.
_root
)
return
'<FileSystem:
%s
- Root Directory:
%s
>'
%
args
__str__
=
__repr__
def
__unicode__
(
self
):
args
=
(
self
.
__class__
.
__name__
,
self
.
_root
)
return
u'<FileSystem:
%s
- Root Directory:
%s
>'
%
args
def
_update
(
self
,
path
,
data
):
"""
Updates contents of an existing file
@param path: Id of the file for which to update content
@param data: Contents to write to the file
@return: Id of the updated file
"""
path
=
self
.
_normpath
(
path
)
if
isinstance
(
data
,
six
.
string_types
):
string_data
=
data
else
:
try
:
data
.
seek
(
0
)
string_data
=
data
.
read
()
except
:
raise
ResourceInvalidError
(
"Unsupported type"
)
f
=
self
.
getinfo
(
path
)
return
self
.
client
.
put
(
(
f
[
"title"
],
string_data
),
f
[
"parent_id"
],
True
)[
'id'
]
def
setcontents
(
self
,
path
,
data
=
""
,
chunk_size
=
64
*
1024
,
**
kwargs
):
"""
Sets new content to remote file
Method works only with existing files and sets new content to them.
@param path: Id of the file in which to write the new content
@param contents: File contents as a string, or any object with read
and seek methods
@param kwargs: additional parameters like:
encoding: the type of encoding to use if data is text
errors: encoding errors
@param chunk_size: Number of bytes to read in a chunk, if the
implementation has to resort to a read / copy loop.
@return: Id of the updated file
"""
path
=
self
.
_normpath
(
path
)
encoding
=
kwargs
.
get
(
"encoding"
,
None
)
errors
=
kwargs
.
get
(
"errors"
,
None
)
if
isinstance
(
data
,
six
.
text_type
):
data
=
data
.
encode
(
encoding
=
encoding
,
errors
=
errors
)
return
self
.
_update
(
path
,
data
)
def
createfile
(
self
,
path
,
wipe
=
True
,
**
kwargs
):
"""
Creates always an empty file, even if another file with the same name
exists.
@param path: path to the new file. It has to be in one of following forms:
- parent_id/file_title.ext
- file_title.ext or /file_title.ext - In this cases root directory is the parent
@param wipe: New file with empty content. If a file with the same name exists it will be
overwritten.
@raise ResourceNotFoundError: If parent doesn't exist.
@attention: Root directory is the current root directory of this instance of
filesystem and not the root of your Google Drive.
@return: Id of the created file
"""
parts
=
path
.
split
(
"/"
)
if
parts
[
0
]
==
""
:
parent_id
=
self
.
_root
title
=
parts
[
1
]
elif
len
(
parts
)
==
2
:
parent_id
=
parts
[
0
]
title
=
parts
[
1
]
if
not
self
.
exists
(
parent_id
):
raise
ResourceNotFoundError
(
"parent doesn't exist"
)
else
:
parent_id
=
self
.
_root
title
=
parts
[
0
]
return
self
.
client
.
put_file
(
parent_id
,
title
,
""
,
wipe
)
def
open
(
self
,
path
,
mode
=
'r'
,
buffering
=-
1
,
encoding
=
None
,
errors
=
None
,
newline
=
None
,
line_buffering
=
False
,
**
kwargs
):
"""
Open the named file in the given mode.
This method downloads the file contents into a local temporary file
so that it can be worked on efficiently. Any changes made to the
file are only sent back to cloud storage when the file is flushed or closed.
@param path: Id of the file to be opened
@param mode: In which mode to open the file
@raise ResourceNotFoundError: If given path doesn't exist and 'w' is not in mode
@return: RemoteFileBuffer object
"""
path
=
self
.
_normpath
(
path
)
spooled_file
=
SpooledTemporaryFile
(
mode
=
mode
,
bufsize
=
MAX_BUFFER
)
# Truncate the file if requested
if
"w"
in
mode
:
try
:
self
.
_update
(
path
,
""
)
except
:
path
=
self
.
createfile
(
path
,
True
)
else
:
try
:
spooled_file
.
write
(
self
.
client
.
get_file
(
path
))
spooled_file
.
seek
(
0
,
0
)
except
Exception
as
e
:
if
"w"
not
in
mode
and
"a"
not
in
mode
:
raise
ResourceNotFoundError
(
"
%r
"
%
e
)
else
:
path
=
self
.
createfile
(
path
,
True
)
return
RemoteFileBuffer
(
self
,
path
,
mode
,
spooled_file
)
def
is_root
(
self
,
path
):
path
=
self
.
_normpath
(
path
)
if
path
==
self
.
_root
:
return
True
else
:
return
False
def
copy
(
self
,
src
,
dst
,
overwrite
=
False
,
chunk_size
=
1024
*
64
):
"""
@param src: Id of the file to be copied
@param dst: Id of the folder in which to copy the file
@param overwrite: This is never true for OneDrive
@return: Id of the copied file
"""
return
self
.
client
.
file_copy
(
src
,
dst
)
def
copydir
(
self
,
src
,
dst
,
overwrite
=
False
,
ignore_errors
=
False
,
chunk_size
=
16384
):
"""
NOTE: OneDrive doesn't support copy of folders. And to implement it
over copy method will be very inefficient
"""
raise
NotImplemented
(
"If implemented method will be very inefficient"
)
def
rename
(
self
,
src
,
dst
):
"""
@param src: Id of the file to be renamed
@param dst: New title of the file
@raise UnsupportedError: If trying to rename the root directory
@return: Id of the renamed file
"""
if
self
.
is_root
(
path
=
src
):
raise
UnsupportedError
(
"Can't rename the root directory"
)
return
self
.
client
.
update_file
(
src
,
{
"name"
:
dst
})
def
remove
(
self
,
path
):
"""
@param path: id of the folder to be deleted
@return: None if removal was successful
"""
path
=
self
.
_normpath
(
path
)
if
self
.
is_root
(
path
=
path
):
raise
UnsupportedError
(
"Can't remove the root directory"
)
if
self
.
isdir
(
path
=
path
):
raise
ResourceInvalidError
(
"Specified path is a directory"
)
return
self
.
client
.
file_delete
(
path
)
def
removedir
(
self
,
path
):
"""
@param path: id of the folder to be deleted
@return: None if removal was successful
"""
path
=
self
.
_normpath
(
path
)
if
not
self
.
isdir
(
path
):
raise
ResourceInvalidError
(
"Specified path is a directory"
)
if
self
.
is_root
(
path
=
path
):
raise
UnsupportedError
(
"remove the root directory"
)
return
self
.
client
.
file_delete
(
path
)
def
makedir
(
self
,
path
,
recursive
=
False
,
allow_recreate
=
False
):
"""
@param path: path to the folder you want to create.
it has to be in one of the following forms:
- parent_id/new_folder_name (when recursive is False)
- parent_id/new_folder1/new_folder2... (when recursive is True)
- /new_folder_name to create a new folder in root directory
- /new_folder1/new_folder2... to recursively create a new folder in root
@param recursive: allows recursive creation of directories
@param allow_recreate: for OneDrive this param is always False, it will
never recreate a directory
"""
parts
=
path
.
split
(
"/"
)
if
parts
[
0
]
==
""
:
parent_id
=
self
.
_root
elif
len
(
parts
)
>=
2
:
parent_id
=
parts
[
0
]
if
not
self
.
exists
(
parent_id
):
raise
ResourceNotFoundError
(
"parent with the id '
%s
' doesn't exist"
%
parent_id
)
if
len
(
parts
)
>
2
:
if
recursive
:
for
i
in
range
(
len
(
parts
)
-
1
):
title
=
parts
[
i
+
1
]
resp
=
self
.
client
.
file_create_folder
(
parent_id
,
title
)
parent_id
=
resp
else
:
raise
UnsupportedError
(
"recursively create a folder"
)
return
resp
else
:
if
len
(
parts
)
==
1
:
title
=
parts
[
0
]
parent_id
=
self
.
_root
else
:
title
=
parts
[
1
]
return
self
.
client
.
file_create_folder
(
parent_id
,
title
)
def
move
(
self
,
src
,
dst
,
overwrite
=
False
,
chunk_size
=
16384
):
"""
@param src: id of the file to be moved
@param dst: id of the folder in which the file will be moved
@param overwrite: for Sky drive it is always false
@param chunk_size: if using chunk upload
@return: Id of the moved file
"""
if
self
.
isdir
(
src
):
raise
UnsupportedError
(
"move a directory"
)
return
self
.
client
.
file_move
(
src
,
dst
)
def
movedir
(
self
,
src
,
dst
,
overwrite
=
False
,
ignore_errors
=
False
,
chunk_size
=
16384
):
"""
@attention: onedrive API doesn't allow to move folders
"""
raise
UnsupportedError
(
"move a directory"
)
def
isdir
(
self
,
path
):
"""
Checks if the given path is a folder
@param path: id of the object to check
@attention: this method doesn't check if the given path exists
it will return true or false even if the file/folder doesn't exist
"""
path
=
self
.
_normpath
(
path
)
info
=
self
.
client
.
info
(
path
)
return
"folder"
in
path
or
info
[
'type'
]
==
"folder"
def
isfile
(
self
,
path
):
"""
Checks if the given path is a file
@param path: id of the object to check
@attention: this method doesn't check if the given path exists
it will return true or false even if the file/folder doesn't exist
"""
path
=
self
.
_normpath
(
path
)
info
=
self
.
client
.
info
(
path
)
return
"file"
in
path
or
info
[
'type'
]
==
"file"
def
exists
(
self
,
path
):
"""
Checks if a the specified path exists
@param path: Id of the file/folder to check
"""
try
:
return
self
.
client
.
metadata
(
path
)
except
RemoteConnectionError
as
e
:
raise
RemoteConnectionError
(
e
)
except
:
return
False
def
listdir
(
self
,
path
=
None
,
wildcard
=
None
,
full
=
False
,
absolute
=
False
,
dirs_only
=
False
,
files_only
=
False
,
overrideCache
=
False
):
"""
Lists the the files and directories under a given path.
The directory contents are returned as a list of unicode paths.
@param path: id of the folder to list
@type path: string
@param wildcard: Only returns paths that match this wildcard
@type wildcard: string containing a wildcard, or a callable that accepts a path and returns a boolean
@param full: returns full paths (relative to the root)
@type full: bool
@param absolute: returns absolute paths (paths beginning with /)
@type absolute: bool
@param dirs_only: if True, only return directories
@type dirs_only: bool
@param files_only: if True, only return files
@type files_only: bool
@return: a list of unicode paths
"""
path
=
self
.
_normpath
(
path
)
flist
=
self
.
client
.
children
(
path
)
dirContent
=
self
.
_listdir_helper
(
''
,
flist
,
wildcard
,
full
,
absolute
,
dirs_only
,
files_only
)
return
dirContent
def
listdirinfo
(
self
,
path
=
None
,
wildcard
=
None
,
full
=
False
,
absolute
=
False
,
dirs_only
=
False
,
files_only
=
False
):
"""
Retrieves a list of paths and path info under a given path.
This method behaves like listdir() but instead of just returning
the name of each item in the directory, it returns a tuple of the
name and the info dict as returned by getinfo.
@param path: id of the folder
@param wildcard: filter paths that match this wildcard
@param dirs_only: only retrieve directories
@type dirs_only: bool
@param files_only: only retrieve files
@type files_only: bool
@return: tuple of the name and the info dict as returned by getinfo.
"""
#Optimised listdir from pyfs
return
[(
p
,
self
.
getinfo
(
p
))
for
p
in
self
.
listdir
(
path
,
wildcard
=
wildcard
,
full
=
full
,
absolute
=
absolute
,
dirs_only
=
dirs_only
,
files_only
=
files_only
)]
def
getinfo
(
self
,
path
):
"""
Returned information is metadata from cloud service +
a few more fields with standard names for some parts
of the metadata.
@param path: file id for which to return informations
@return: dictionary with informations about the specific file
"""
path
=
self
.
_normpath
(
path
)
# Fix for onedrive, because if you don't write the path in the way they
# want. It will raise an exception with the error code 400.
# That error means nothing because it's raised in many situations.
if
not
self
.
exists
(
path
):
raise
ResourceNotFoundError
(
path
=
path
)
return
self
.
_metadata_to_info
(
self
.
client
.
metadata
(
path
))
def
getpathurl
(
self
,
path
,
allow_none
=
False
):
"""Returns a url that corresponds to the given path, if one exists.
If the path does not have an equivalent URL form (and allow_none is False)
then a :class:`~fs.errors.NoPathURLError` exception is thrown. Otherwise the URL will be
returns as an unicode string.
@param path: object id for which to return url path
@param allow_none: if true, this method can return None if there is no
URL form of the given path
@type allow_none: bool
@raises `fs.errors.NoPathURLError`: If no URL form exists, and allow_none is False (the default)
@rtype: unicode
"""
url
=
None
try
:
url
=
self
.
getinfo
(
path
)[
'source'
]
except
RemoteConnectionError
as
e
:
raise
RemoteConnectionError
(
e
)
except
:
if
not
allow_none
:
raise
NoPathURLError
(
path
=
path
)
return
url
def
desc
(
self
,
path
):
"""
@return: The title for the given path.
"""
info
=
self
.
getinfo
(
path
)
return
info
[
"title"
]
def
about
(
self
):
"""This methods returns information about the current user with whose
credentials is the file system instantiated."""
user
=
self
.
client
.
info
(
"me"
)
info
=
{}
quota
=
self
.
client
.
get_quota
()
info
[
'cloud_storage_url'
]
=
"https://onedrive.live.com/"
info
[
'user_name'
]
=
user
[
'first_name'
]
.
capitalize
()
+
\
" "
+
user
[
'last_name'
]
.
capitalize
()
info
[
'quota'
]
=
100
*
(
float
(
quota
[
1
]
-
quota
[
0
])
/
float
(
quota
[
1
]))
return
info
def
_normpath
(
self
,
path
):
"""
Method normalizes the path for sky drive.
@return: normalized path as a string
"""
if
path
in
(
"/skydrive/camera_roll"
,
"/skydrive/my_documents"
,
"/skydrive/my_photos"
,
"/skydrive/public_documents"
,
"me/skydrive"
):
return
path
elif
path
==
self
.
_root
:
return
path
elif
path
is
None
or
path
==
""
:
return
self
.
_root
elif
len
(
path
.
split
(
"/"
))
>
2
:
return
path
.
split
(
"/"
)[
-
1
]
elif
path
[
0
]
==
"/"
and
len
(
path
)
==
1
:
return
self
.
_root
elif
path
[
0
]
==
"/"
:
return
path
[
1
:]
elif
len
(
path
)
==
0
:
return
self
.
_root
return
path
def
_metadata_to_info
(
self
,
metadata
,
localtime
=
False
):
"""
Method adds a few standard names to the metadata:
size - the size of the file/folder
isdir - is something a file or a directory
created_time - the time of the creation
path - path to the object which metadata are we showing
revision - sky drive doesn't have a revision parameter
modified - time of the last modification
@return: The full metadata and a few more fields
with standard names.
"""
path
=
metadata
.
get
(
'id'
,
'0'
)
isdir
=
"folder"
in
path
or
metadata
.
get
(
'type'
)
==
"folder"
info
=
{
'isdir'
:
isdir
,
'created_time'
:
metadata
.
get
(
'createdDate'
,
0
),
'title'
:
metadata
.
get
(
'name'
,
0
),
'path'
:
path
,
'revision'
:
None
,
'created_time'
:
metadata
.
get
(
'createdDate'
,
0
),
'modified'
:
metadata
.
get
(
"updated_time"
,
0
)
}
info
.
update
(
metadata
)
return
info
Event Timeline
Log In to Comment