class SharePointClient_v1
Comprehensive SharePoint client using app-only authentication. Requires Azure AD app registration with appropriate SharePoint permissions.
/tf/active/vicechatdev/Sharepoint/sharepoint_client.py
11 - 540
moderate
Purpose
Comprehensive SharePoint client using app-only authentication. Requires Azure AD app registration with appropriate SharePoint permissions.
Source Code
class SharePointClient:
"""
Comprehensive SharePoint client using app-only authentication.
Requires Azure AD app registration with appropriate SharePoint permissions.
"""
def __init__(self, site_url: str, client_id: str, client_secret: str):
"""
Initialize SharePoint client with app-only authentication.
Args:
site_url: SharePoint site URL (e.g., https://tenant.sharepoint.com/sites/sitename)
client_id: Azure AD app client ID
client_secret: Azure AD app client secret
"""
self.site_url = site_url
self.client_id = client_id
self.client_secret = client_secret
# Initialize context with app-only authentication
self.ctx = ClientContext(site_url).with_credentials(
ClientCredential(client_id, client_secret)
)
# Setup logging
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
# Test connection
self._test_connection()
def _test_connection(self) -> bool:
"""Test the SharePoint connection."""
try:
web = self.ctx.web
self.ctx.load(web)
self.ctx.execute_query()
self.logger.info(f"Successfully connected to SharePoint site: {web.properties['Title']}")
return True
except Exception as e:
self.logger.error(f"Failed to connect to SharePoint: {e}")
raise ConnectionError(f"SharePoint connection failed: {e}")
def list_files(self, folder_path: str = "/Shared Documents",
include_subfolders: bool = False) -> List[Dict]:
"""
List all files in a SharePoint folder.
Args:
folder_path: Server relative path to folder
include_subfolders: Whether to include files from subfolders
Returns:
List of file information dictionaries
"""
try:
folder = self.ctx.web.get_folder_by_server_relative_url(folder_path)
files = folder.files
self.ctx.load(files)
self.ctx.execute_query()
file_list = []
for file in files:
file_info = {
'name': file.properties['Name'],
'server_relative_url': file.properties['ServerRelativeUrl'],
'size': file.properties['Length'],
'modified': file.properties['TimeLastModified'],
'created': file.properties['TimeCreated'],
'author': file.properties.get('Author', {}).get('Title', 'Unknown'),
'file_type': file.properties['Name'].split('.')[-1] if '.' in file.properties['Name'] else ''
}
file_list.append(file_info)
if include_subfolders:
subfolders = folder.folders
self.ctx.load(subfolders)
self.ctx.execute_query()
for subfolder in subfolders:
if not subfolder.properties['Name'].startswith('.'): # Skip system folders
subfolder_path = subfolder.properties['ServerRelativeUrl']
subfolder_files = self.list_files(subfolder_path, include_subfolders=True)
file_list.extend(subfolder_files)
self.logger.info(f"Listed {len(file_list)} files from {folder_path}")
return file_list
except Exception as e:
self.logger.error(f"Error listing files in {folder_path}: {e}")
raise
def list_folders(self, folder_path: str = "/Shared Documents") -> List[Dict]:
"""
List all folders in a SharePoint folder.
Args:
folder_path: Server relative path to folder
Returns:
List of folder information dictionaries
"""
try:
folder = self.ctx.web.get_folder_by_server_relative_url(folder_path)
folders = folder.folders
self.ctx.load(folders)
self.ctx.execute_query()
folder_list = []
for subfolder in folders:
if not subfolder.properties['Name'].startswith('.'): # Skip system folders
folder_info = {
'name': subfolder.properties['Name'],
'server_relative_url': subfolder.properties['ServerRelativeUrl'],
'item_count': subfolder.properties.get('ItemCount', 0),
'created': subfolder.properties.get('TimeCreated'),
'modified': subfolder.properties.get('TimeLastModified')
}
folder_list.append(folder_info)
self.logger.info(f"Listed {len(folder_list)} folders from {folder_path}")
return folder_list
except Exception as e:
self.logger.error(f"Error listing folders in {folder_path}: {e}")
raise
def download_file(self, server_relative_url: str, local_path: str,
create_dirs: bool = True) -> bool:
"""
Download a file from SharePoint.
Args:
server_relative_url: SharePoint server relative URL of file
local_path: Local file path to save to
create_dirs: Whether to create local directories if they don't exist
Returns:
True if successful, False otherwise
"""
try:
if create_dirs:
os.makedirs(os.path.dirname(local_path), exist_ok=True)
file = self.ctx.web.get_file_by_server_relative_url(server_relative_url)
with open(local_path, "wb") as local_file:
file.download(local_file)
self.ctx.execute_query()
file_size = os.path.getsize(local_path)
self.logger.info(f"Downloaded {server_relative_url} to {local_path} ({file_size} bytes)")
return True
except Exception as e:
self.logger.error(f"Error downloading file {server_relative_url}: {e}")
return False
def upload_file(self, local_file_path: str, sharepoint_folder_path: str,
overwrite: bool = True, chunk_size: int = 1024*1024,
checkout_before_upload: bool = False,
checkin_comment: str = "Uploaded via Python") -> Optional[Dict]:
"""
Upload a file to SharePoint with enhanced version control.
Args:
local_file_path: Path to local file
sharepoint_folder_path: SharePoint folder path
overwrite: Whether to overwrite existing files
chunk_size: Upload chunk size for large files (default 1MB)
checkout_before_upload: Whether to checkout file before upload (recommended for versioning)
checkin_comment: Comment for the checkin operation
Returns:
File information dictionary if successful, None otherwise
"""
try:
if not os.path.exists(local_file_path):
raise FileNotFoundError(f"Local file not found: {local_file_path}")
file_name = os.path.basename(local_file_path)
file_size = os.path.getsize(local_file_path)
target_folder = self.ctx.web.get_folder_by_server_relative_url(sharepoint_folder_path)
file_url = f"{sharepoint_folder_path}/{file_name}"
# Check if file exists and handle checkout if requested
existing_file = None
try:
existing_file = self.ctx.web.get_file_by_server_relative_url(file_url)
self.ctx.load(existing_file)
self.ctx.execute_query()
if checkout_before_upload:
# Check out the file before uploading
existing_file.check_out()
self.ctx.execute_query()
self.logger.info(f"Checked out file: {file_name}")
except Exception:
# File doesn't exist, which is fine
existing_file = None
# Upload the file
if file_size > chunk_size:
self.logger.info(f"Uploading large file {file_name} ({file_size} bytes) in chunks")
target_file = target_folder.files.create_upload_session(
local_file_path, chunk_size, overwrite
).execute_query()
else:
with open(local_file_path, 'rb') as content_file:
file_content = content_file.read()
if overwrite:
target_file = target_folder.upload_file(file_name, file_content)
else:
target_file = target_folder.files.add(file_name, file_content)
self.ctx.execute_query()
# Check in the file if it was checked out
if existing_file and checkout_before_upload:
target_file.checkin(checkin_comment)
self.ctx.execute_query()
self.logger.info(f"Checked in file: {file_name} with comment: {checkin_comment}")
# Get file information including version
self.ctx.load(target_file)
self.ctx.execute_query()
file_info = {
'name': target_file.properties['Name'],
'server_relative_url': target_file.properties['ServerRelativeUrl'],
'size': target_file.properties['Length'],
'version': target_file.properties.get('UIVersionLabel', '1.0'),
'upload_time': datetime.now().isoformat(),
'is_new_file': existing_file is None
}
self.logger.info(f"Uploaded {local_file_path} to {sharepoint_folder_path}/{file_name} (Version: {file_info['version']})")
return file_info
except Exception as e:
self.logger.error(f"Error uploading file {local_file_path}: {e}")
return None
def delete_file(self, server_relative_url: str) -> bool:
"""
Delete a file from SharePoint.
Args:
server_relative_url: SharePoint server relative URL of file
Returns:
True if successful, False otherwise
"""
try:
file = self.ctx.web.get_file_by_server_relative_url(server_relative_url)
file.delete_object()
self.ctx.execute_query()
self.logger.info(f"Deleted file: {server_relative_url}")
return True
except Exception as e:
self.logger.error(f"Error deleting file {server_relative_url}: {e}")
return False
def create_folder(self, parent_folder_path: str, folder_name: str) -> Optional[Dict]:
"""
Create a new folder in SharePoint.
Args:
parent_folder_path: Path to parent folder
folder_name: Name of new folder
Returns:
Folder information dictionary if successful, None otherwise
"""
try:
parent_folder = self.ctx.web.get_folder_by_server_relative_url(parent_folder_path)
new_folder = parent_folder.folders.add(folder_name)
self.ctx.execute_query()
folder_info = {
'name': new_folder.properties['Name'],
'server_relative_url': new_folder.properties['ServerRelativeUrl'],
'created': datetime.now().isoformat()
}
self.logger.info(f"Created folder: {parent_folder_path}/{folder_name}")
return folder_info
except Exception as e:
self.logger.error(f"Error creating folder {folder_name} in {parent_folder_path}: {e}")
return None
def update_file_metadata(self, server_relative_url: str, metadata: Dict) -> bool:
"""
Update file metadata/properties.
Args:
server_relative_url: SharePoint server relative URL of file
metadata: Dictionary of metadata to update
Returns:
True if successful, False otherwise
"""
try:
file = self.ctx.web.get_file_by_server_relative_url(server_relative_url)
list_item = file.listItemAllFields
for key, value in metadata.items():
list_item.set_property(key, value)
list_item.update()
self.ctx.execute_query()
self.logger.info(f"Updated metadata for: {server_relative_url}")
return True
except Exception as e:
self.logger.error(f"Error updating metadata for {server_relative_url}: {e}")
return False
def get_file_info(self, server_relative_url: str) -> Optional[Dict]:
"""
Get detailed information about a file.
Args:
server_relative_url: SharePoint server relative URL of file
Returns:
File information dictionary if successful, None otherwise
"""
try:
file = self.ctx.web.get_file_by_server_relative_url(server_relative_url)
self.ctx.load(file)
self.ctx.execute_query()
# Get list item for additional metadata
list_item = file.listItemAllFields
self.ctx.load(list_item)
self.ctx.execute_query()
file_info = {
'name': file.properties['Name'],
'server_relative_url': file.properties['ServerRelativeUrl'],
'size': file.properties['Length'],
'modified': file.properties['TimeLastModified'],
'created': file.properties['TimeCreated'],
'version': file.properties.get('UIVersionLabel', '1.0'),
'check_out_type': file.properties.get('CheckOutType', 0),
'content_type': list_item.properties.get('ContentType', {}).get('Name', 'Unknown'),
'metadata': {k: v for k, v in list_item.properties.items()
if not k.startswith('_') and k not in ['ContentType', 'File']}
}
return file_info
except Exception as e:
self.logger.error(f"Error getting file info for {server_relative_url}: {e}")
return None
def copy_file(self, source_url: str, destination_url: str, overwrite: bool = True) -> bool:
"""
Copy a file within SharePoint.
Args:
source_url: Source file server relative URL
destination_url: Destination server relative URL
overwrite: Whether to overwrite existing files
Returns:
True if successful, False otherwise
"""
try:
source_file = self.ctx.web.get_file_by_server_relative_url(source_url)
source_file.copyto(destination_url, overwrite)
self.ctx.execute_query()
self.logger.info(f"Copied file from {source_url} to {destination_url}")
return True
except Exception as e:
self.logger.error(f"Error copying file from {source_url} to {destination_url}: {e}")
return False
def move_file(self, source_url: str, destination_url: str, overwrite: bool = True) -> bool:
"""
Move a file within SharePoint.
Args:
source_url: Source file server relative URL
destination_url: Destination server relative URL
overwrite: Whether to overwrite existing files
Returns:
True if successful, False otherwise
"""
try:
source_file = self.ctx.web.get_file_by_server_relative_url(source_url)
source_file.moveto(destination_url, overwrite)
self.ctx.execute_query()
self.logger.info(f"Moved file from {source_url} to {destination_url}")
return True
except Exception as e:
self.logger.error(f"Error moving file from {source_url} to {destination_url}: {e}")
return False
def search_files(self, folder_path: str, search_term: str,
include_subfolders: bool = True) -> List[Dict]:
"""
Search for files by name in a folder.
Args:
folder_path: Folder path to search in
search_term: Term to search for in file names
include_subfolders: Whether to search in subfolders
Returns:
List of matching file information dictionaries
"""
all_files = self.list_files(folder_path, include_subfolders)
matching_files = [
file_info for file_info in all_files
if search_term.lower() in file_info['name'].lower()
]
self.logger.info(f"Found {len(matching_files)} files matching '{search_term}'")
return matching_files
def get_file_versions(self, server_relative_url: str) -> List[Dict]:
"""
Get all versions of a file.
Args:
server_relative_url: SharePoint server relative URL of file
Returns:
List of version information dictionaries
"""
try:
file = self.ctx.web.get_file_by_server_relative_url(server_relative_url)
versions = file.versions
self.ctx.load(versions)
self.ctx.execute_query()
version_list = []
for version in versions:
version_info = {
'version_label': version.properties['VersionLabel'],
'created': version.properties['Created'],
'created_by': version.properties.get('CreatedBy', {}).get('Title', 'Unknown'),
'size': version.properties['Size'],
'url': version.properties['Url'],
'is_current': version.properties.get('IsCurrentVersion', False)
}
version_list.append(version_info)
# Add current version info
self.ctx.load(file)
self.ctx.execute_query()
current_version = {
'version_label': file.properties.get('UIVersionLabel', '1.0'),
'created': file.properties['TimeLastModified'],
'size': file.properties['Length'],
'url': file.properties['ServerRelativeUrl'],
'is_current': True
}
version_list.append(current_version)
self.logger.info(f"Found {len(version_list)} versions for {server_relative_url}")
return version_list
except Exception as e:
self.logger.error(f"Error getting versions for {server_relative_url}: {e}")
return []
def download_file_version(self, server_relative_url: str, version_label: str,
local_path: str, create_dirs: bool = True) -> bool:
"""
Download a specific version of a file.
Args:
server_relative_url: SharePoint server relative URL of file
version_label: Version to download (e.g., "1.0", "2.0")
local_path: Local file path to save to
create_dirs: Whether to create local directories
Returns:
True if successful, False otherwise
"""
try:
if create_dirs:
os.makedirs(os.path.dirname(local_path), exist_ok=True)
file = self.ctx.web.get_file_by_server_relative_url(server_relative_url)
versions = file.versions
self.ctx.load(versions)
self.ctx.execute_query()
target_version = None
for version in versions:
if version.properties['VersionLabel'] == version_label:
target_version = version
break
if not target_version:
self.logger.error(f"Version {version_label} not found for {server_relative_url}")
return False
# Download the specific version
version_url = target_version.properties['Url']
version_file = self.ctx.web.get_file_by_server_relative_url(version_url)
with open(local_path, "wb") as local_file:
version_file.download(local_file)
self.ctx.execute_query()
file_size = os.path.getsize(local_path)
self.logger.info(f"Downloaded version {version_label} of {server_relative_url} to {local_path} ({file_size} bytes)")
return True
except Exception as e:
self.logger.error(f"Error downloading version {version_label} of {server_relative_url}: {e}")
return False
Parameters
| Name | Type | Default | Kind |
|---|---|---|---|
bases |
- | - |
Parameter Details
bases: Parameter of type
Return Value
Returns unspecified type
Class Interface
Methods
__init__(self, site_url, client_id, client_secret)
Purpose: Initialize SharePoint client with app-only authentication. Args: site_url: SharePoint site URL (e.g., https://tenant.sharepoint.com/sites/sitename) client_id: Azure AD app client ID client_secret: Azure AD app client secret
Parameters:
site_url: Type: strclient_id: Type: strclient_secret: Type: str
Returns: None
_test_connection(self) -> bool
Purpose: Test the SharePoint connection.
Returns: Returns bool
list_files(self, folder_path, include_subfolders) -> List[Dict]
Purpose: List all files in a SharePoint folder. Args: folder_path: Server relative path to folder include_subfolders: Whether to include files from subfolders Returns: List of file information dictionaries
Parameters:
folder_path: Type: strinclude_subfolders: Type: bool
Returns: Returns List[Dict]
list_folders(self, folder_path) -> List[Dict]
Purpose: List all folders in a SharePoint folder. Args: folder_path: Server relative path to folder Returns: List of folder information dictionaries
Parameters:
folder_path: Type: str
Returns: Returns List[Dict]
download_file(self, server_relative_url, local_path, create_dirs) -> bool
Purpose: Download a file from SharePoint. Args: server_relative_url: SharePoint server relative URL of file local_path: Local file path to save to create_dirs: Whether to create local directories if they don't exist Returns: True if successful, False otherwise
Parameters:
server_relative_url: Type: strlocal_path: Type: strcreate_dirs: Type: bool
Returns: Returns bool
upload_file(self, local_file_path, sharepoint_folder_path, overwrite, chunk_size, checkout_before_upload, checkin_comment) -> Optional[Dict]
Purpose: Upload a file to SharePoint with enhanced version control. Args: local_file_path: Path to local file sharepoint_folder_path: SharePoint folder path overwrite: Whether to overwrite existing files chunk_size: Upload chunk size for large files (default 1MB) checkout_before_upload: Whether to checkout file before upload (recommended for versioning) checkin_comment: Comment for the checkin operation Returns: File information dictionary if successful, None otherwise
Parameters:
local_file_path: Type: strsharepoint_folder_path: Type: stroverwrite: Type: boolchunk_size: Type: intcheckout_before_upload: Type: boolcheckin_comment: Type: str
Returns: Returns Optional[Dict]
delete_file(self, server_relative_url) -> bool
Purpose: Delete a file from SharePoint. Args: server_relative_url: SharePoint server relative URL of file Returns: True if successful, False otherwise
Parameters:
server_relative_url: Type: str
Returns: Returns bool
create_folder(self, parent_folder_path, folder_name) -> Optional[Dict]
Purpose: Create a new folder in SharePoint. Args: parent_folder_path: Path to parent folder folder_name: Name of new folder Returns: Folder information dictionary if successful, None otherwise
Parameters:
parent_folder_path: Type: strfolder_name: Type: str
Returns: Returns Optional[Dict]
update_file_metadata(self, server_relative_url, metadata) -> bool
Purpose: Update file metadata/properties. Args: server_relative_url: SharePoint server relative URL of file metadata: Dictionary of metadata to update Returns: True if successful, False otherwise
Parameters:
server_relative_url: Type: strmetadata: Type: Dict
Returns: Returns bool
get_file_info(self, server_relative_url) -> Optional[Dict]
Purpose: Get detailed information about a file. Args: server_relative_url: SharePoint server relative URL of file Returns: File information dictionary if successful, None otherwise
Parameters:
server_relative_url: Type: str
Returns: Returns Optional[Dict]
copy_file(self, source_url, destination_url, overwrite) -> bool
Purpose: Copy a file within SharePoint. Args: source_url: Source file server relative URL destination_url: Destination server relative URL overwrite: Whether to overwrite existing files Returns: True if successful, False otherwise
Parameters:
source_url: Type: strdestination_url: Type: stroverwrite: Type: bool
Returns: Returns bool
move_file(self, source_url, destination_url, overwrite) -> bool
Purpose: Move a file within SharePoint. Args: source_url: Source file server relative URL destination_url: Destination server relative URL overwrite: Whether to overwrite existing files Returns: True if successful, False otherwise
Parameters:
source_url: Type: strdestination_url: Type: stroverwrite: Type: bool
Returns: Returns bool
search_files(self, folder_path, search_term, include_subfolders) -> List[Dict]
Purpose: Search for files by name in a folder. Args: folder_path: Folder path to search in search_term: Term to search for in file names include_subfolders: Whether to search in subfolders Returns: List of matching file information dictionaries
Parameters:
folder_path: Type: strsearch_term: Type: strinclude_subfolders: Type: bool
Returns: Returns List[Dict]
get_file_versions(self, server_relative_url) -> List[Dict]
Purpose: Get all versions of a file. Args: server_relative_url: SharePoint server relative URL of file Returns: List of version information dictionaries
Parameters:
server_relative_url: Type: str
Returns: Returns List[Dict]
download_file_version(self, server_relative_url, version_label, local_path, create_dirs) -> bool
Purpose: Download a specific version of a file. Args: server_relative_url: SharePoint server relative URL of file version_label: Version to download (e.g., "1.0", "2.0") local_path: Local file path to save to create_dirs: Whether to create local directories Returns: True if successful, False otherwise
Parameters:
server_relative_url: Type: strversion_label: Type: strlocal_path: Type: strcreate_dirs: Type: bool
Returns: Returns bool
Required Imports
from office365.sharepoint.client_context import ClientContext
from office365.runtime.auth.client_credential import ClientCredential
from office365.sharepoint.files.file import File
from office365.sharepoint.folders.folder import Folder
import os
Usage Example
# Example usage:
# result = SharePointClient(bases)
Tags
Similar Components
AI-powered semantic similarity - components with related functionality:
-
class SharePointRestClient 70.9% similar
-
class SharePointClient 70.5% similar
-
class SharePointGraphClient 68.2% similar
-
function main_v35 61.7% similar
-
function provide_admin_instructions 58.3% similar